mecha10-cli 0.8.0

//! Spawning nodes by name for dev mode

use super::models::{ensure_node_models_available, ensure_ollama_models_available};
use crate::context::CliContext;
use crate::services::CredentialsService;
use crate::types::{turn_config, ProjectConfig, TurnConfig};
use crate::ui::LogBuffer;
use anyhow::Result;
use serde::Deserialize;
use std::collections::HashMap;

pub const SIM_VISIBLE_ENV_VAR: &str = "MECHA10_DEV_SIM_VISIBLE";

const SIMULATOR_VISIBLE_ENV_VAR: &str = "MECHA10_SIMULATOR_VISIBLE";

const SIMULATOR_MODEL_ENV_VAR: &str = "MECHA10_SIMULATOR_MODEL";
const SIMULATOR_ENV_ENV_VAR: &str = "MECHA10_SIMULATOR_ENV";
const SIMULATOR_PRESET_ENV_VAR: &str = "MECHA10_SIMULATOR_PRESET";

const TURN_URL_ENV_VAR: &str = "MECHA10_WEBRTC_TURN_URL";
const TURN_USERNAME_ENV_VAR: &str = "MECHA10_WEBRTC_TURN_USERNAME";
const TURN_CREDENTIAL_ENV_VAR: &str = "MECHA10_WEBRTC_TURN_CREDENTIAL";

#[derive(Debug, Deserialize)]
struct TurnCredentialsResponse {
    url: String,
    username: String,
    credential: String,
}

/// Node names that actually consume `WEBRTC_RELAY_URL`/`MECHA10_WEBRTC_TURN_*` - see
/// `packages/video/src/webrtc/connection.rs`. Gates the TURN credentials fetch below so
/// `mecha10 dev` doesn't make a network round-trip to mecha10-api on every invocation needlessly.
fn batch_needs_turn_credentials(node_names: &[String]) -> bool {
    node_names
        .iter()
        .any(|n| matches!(super::strip_node_scope(n), "camera" | "video-streamer"))
}

/// `Ok(Some(_))` on a well-formed 200 response, `Ok(None)` on 404 (TURN not configured
/// server-side, not an error), `Err` for anything else (network failure, auth failure,
/// non-2xx/404 status, unexpected body shape).
async fn fetch_turn_credentials(control_plane_url: &str, api_key: &str) -> Result<Option<TurnCredentialsResponse>> {
    let url = format!(
        "{}/api/webrtc/turn-credentials",
        control_plane_url.trim_end_matches('/')
    );
    let response = reqwest::Client::new().get(url).bearer_auth(api_key).send().await?;

    if response.status() == reqwest::StatusCode::NOT_FOUND {
        return Ok(None);
    }

    let creds = response.error_for_status()?.json::<TurnCredentialsResponse>().await?;
    Ok(Some(creds))
}

/// Resolve TURN credentials into `project_env`, highest priority first: an explicit env var
/// override, then a build-time-baked default, then a best-effort runtime fetch from
/// mecha10-api. Never blocks `mecha10 dev` startup; falls back to STUN-only on any failure.
async fn maybe_inject_turn_credentials(
    node_names: &[String],
    control_plane_url: &str,
    api_key: Option<&str>,
    project_env: &mut HashMap<String, String>,
) {
    maybe_inject_turn_credentials_with_baked_resolver(node_names, control_plane_url, api_key, project_env, turn_config)
        .await
}

/// [`maybe_inject_turn_credentials`], but with the baked-default lookup taken as a parameter
/// so tests can stub it instead of relying on real compile-time `option_env!` constants.
async fn maybe_inject_turn_credentials_with_baked_resolver(
    node_names: &[String],
    control_plane_url: &str,
    api_key: Option<&str>,
    project_env: &mut HashMap<String, String>,
    resolve_baked: impl Fn() -> Option<TurnConfig>,
) {
    if std::env::var(TURN_URL_ENV_VAR).is_ok() {
        return;
    }
    if !batch_needs_turn_credentials(node_names) {
        return;
    }

    if let Some(baked) = resolve_baked() {
        project_env.insert(TURN_URL_ENV_VAR.to_string(), baked.url);
        project_env.insert(TURN_USERNAME_ENV_VAR.to_string(), baked.username);
        project_env.insert(TURN_CREDENTIAL_ENV_VAR.to_string(), baked.credential);
        return;
    }

    let Some(api_key) = api_key else {
        return;
    };

    match fetch_turn_credentials(control_plane_url, api_key).await {
        Ok(Some(creds)) => {
            project_env.insert(TURN_URL_ENV_VAR.to_string(), creds.url);
            project_env.insert(TURN_USERNAME_ENV_VAR.to_string(), creds.username);
            project_env.insert(TURN_CREDENTIAL_ENV_VAR.to_string(), creds.credential);
        }
        Ok(None) => {
            println!("\u{2139}\u{fe0f}  TURN server not available, using STUN only");
        }
        Err(e) => {
            tracing::debug!("TURN credentials fetch failed, using STUN only: {}", e);
            println!("\u{2139}\u{fe0f}  TURN server not available, using STUN only");
        }
    }
}

/// When `mecha10 dev --sim-visible` was passed, inject `MECHA10_SIMULATOR_VISIBLE=1` into
/// `project_env` if a `simulator` node is part of this spawn batch; otherwise print a clear
/// no-op message instead of silently doing nothing.
fn maybe_inject_sim_visible_env(node_names: &[String], project_env: &mut HashMap<String, String>) {
    let sim_visible_requested = std::env::var(SIM_VISIBLE_ENV_VAR).is_ok();
    inject_sim_visible_env(sim_visible_requested, node_names, project_env);
}

/// Pure decision logic behind [`maybe_inject_sim_visible_env`], split out so it's unit testable
/// without mutating real process environment variables (which would race across parallel test
/// threads).
fn inject_sim_visible_env(
    sim_visible_requested: bool,
    node_names: &[String],
    project_env: &mut HashMap<String, String>,
) {
    if !sim_visible_requested {
        return;
    }

    let has_simulator = node_names.iter().any(|n| super::strip_node_scope(n) == "simulator");
    if has_simulator {
        project_env.insert(SIMULATOR_VISIBLE_ENV_VAR.to_string(), "1".to_string());
    } else if !node_names.is_empty() {
        println!(
            "\u{2139}\u{fe0f}  --sim-visible has no effect: no `simulator` node in this dev session ({}).",
            node_names.join(", ")
        );
    }
}

/// Inject `mecha10.json`'s `simulation.model`/`simulation.environment`/`simulation.preset` as
/// `MECHA10_SIMULATOR_MODEL`/`MECHA10_SIMULATOR_ENV`/`MECHA10_SIMULATOR_PRESET` when a
/// `simulator` node is part of this spawn batch. Only applies to `mecha10 dev`/`mecha10 run`.
pub(crate) fn inject_simulation_override_env(
    project_config: &ProjectConfig,
    node_names: &[String],
    project_env: &mut HashMap<String, String>,
) {
    let Some(simulation) = project_config.simulation.as_ref() else {
        return;
    };
    if !node_names.iter().any(|n| super::strip_node_scope(n) == "simulator") {
        return;
    }

    if let Some(model) = simulation.model.as_deref().filter(|s| !s.trim().is_empty()) {
        project_env.insert(SIMULATOR_MODEL_ENV_VAR.to_string(), model.to_string());
    }
    if let Some(environment) = simulation.environment.as_deref().filter(|s| !s.trim().is_empty()) {
        project_env.insert(SIMULATOR_ENV_ENV_VAR.to_string(), environment.to_string());
    }
    if let Some(preset) = simulation.preset.as_deref().filter(|s| !s.trim().is_empty()) {
        project_env.insert(SIMULATOR_PRESET_ENV_VAR.to_string(), preset.to_string());
    }
}

/// Inject the project's `actuator.locomotion` config into `project_env` as
/// `MECHA10_LOCOMOTION_MAX_LINEAR_SPEED`/`MECHA10_LOCOMOTION_MAX_ANGULAR_SPEED`/
/// `MECHA10_LOCOMOTION_ACCELERATION`, and its `actuator.joints` config as
/// `MECHA10_ACTUATOR_JOINT_LIMITS`. This is the only place these values come from.
pub(crate) fn inject_locomotion_limits_env(project_config: &ProjectConfig, project_env: &mut HashMap<String, String>) {
    for (key, value) in project_config.actuator.locomotion.as_env_vars() {
        project_env.insert(key.to_string(), value);
    }

    let (key, value) = project_config.actuator.joints_as_env_var();
    project_env.insert(key.to_string(), value);
}

#[derive(Debug, Default)]
pub struct SpawnOutcome {
    pub pids: HashMap<String, u32>,
    pub failures: Vec<(String, String)>,
}

/// Spawn nodes by name. Uses mecha10-node-runner if available (framework dev mode), otherwise
/// spawns nodes directly via the project binary (standalone mode). A per-node spawn failure
/// does not abort the batch - it's recorded in `SpawnOutcome::failures` instead.
pub fn spawn_nodes_by_name(
    ctx: &mut CliContext,
    node_names: &[String],
    project_name: &str,
    project_config: &ProjectConfig,
    log_buffer: Option<&LogBuffer>,
) -> Result<SpawnOutcome> {
    use crate::services::ProcessService;

    println!("Starting nodes...");
    println!();

    let shared_process = ctx.shared_process();
    let mut process_service = shared_process.lock().unwrap();

    let _ = std::fs::OpenOptions::new()
        .create(true)
        .append(true)
        .open("/tmp/command-listener.log")
        .and_then(|mut f| {
            use std::io::Write;
            writeln!(
                f,
                "[spawn_nodes_by_name] Using shared ProcessService, ptr={:p}",
                &*process_service
            )
        });

    let use_node_runner = ProcessService::is_node_runner_available();

    if let Err(e) = tokio::task::block_in_place(|| {
        tokio::runtime::Handle::current().block_on(ensure_node_models_available(node_names))
    }) {
        eprintln!("⚠️  Warning: Failed to check/download models: {}", e);
    }

    if let Err(e) = tokio::task::block_in_place(|| {
        tokio::runtime::Handle::current().block_on(ensure_ollama_models_available(node_names))
    }) {
        eprintln!("⚠️  Warning: Failed to check/pull Ollama models: {}", e);
    }

    if !use_node_runner {
        println!("  (standalone mode - spawning nodes directly)");
        println!();
    }

    let project_env = {
        let credentials_service = CredentialsService::new();
        let api_key = if crate::context::is_ci_environment() {
            match credentials_service.get_api_key() {
                Ok(Some(api_key)) => {
                    tracing::debug!("🔑 Loaded API key from credentials: {}", &api_key[..20]);
                    Some(api_key)
                }
                _ => {
                    tracing::warn!("⚠️  No API key found in credentials - did you run 'mecha10 auth login'?");
                    None
                }
            }
        } else {
            let result = tokio::task::block_in_place(|| {
                tokio::runtime::Handle::current()
                    .block_on(mecha10_auth::ensure_authenticated_interactive(&credentials_service))
            });
            match result {
                Ok(creds) => {
                    tracing::debug!("🔑 Loaded API key from credentials: {}", &creds.api_key[..20]);
                    Some(creds.api_key)
                }
                Err(e) => {
                    tracing::warn!("⚠️  Failed to authenticate: {}", e);
                    None
                }
            }
        };

        let mut env: HashMap<String, String> = mecha10_core::config::robot_process_env(
            &project_config.environments.control_plane_url(),
            &project_config.environments.relay_url(),
            &project_config.environments.redis_url(),
            &project_config.robot.id,
            api_key.as_deref(),
        )
        .into_iter()
        .map(|(k, v)| (k.to_string(), v))
        .collect();

        env.insert("LOG_LEVEL".to_string(), ctx.log_level.to_string().to_lowercase());

        {
            let control_plane_url = project_config.environments.control_plane_url();
            let api_key_for_turn = env.get("ROBOT_API_KEY").cloned();
            tokio::task::block_in_place(|| {
                tokio::runtime::Handle::current().block_on(maybe_inject_turn_credentials(
                    node_names,
                    &control_plane_url,
                    api_key_for_turn.as_deref(),
                    &mut env,
                ))
            });
        }

        maybe_inject_sim_visible_env(node_names, &mut env);

        inject_simulation_override_env(project_config, node_names, &mut env);

        inject_locomotion_limits_env(project_config, &mut env);

        tracing::debug!("📋 Environment variables to inject:");
        tracing::debug!(
            "  CONTROL_PLANE_URL: {}",
            env.get("CONTROL_PLANE_URL").unwrap_or(&"<missing>".to_string())
        );
        tracing::debug!(
            "  ROBOT_ID: {}",
            env.get("ROBOT_ID").unwrap_or(&"<missing>".to_string())
        );
        tracing::debug!(
            "  ROBOT_API_KEY: {}",
            if env.contains_key("ROBOT_API_KEY") {
                "<present>"
            } else {
                "<missing>"
            }
        );
        tracing::debug!(
            "  WEBRTC_RELAY_URL: {}",
            env.get("WEBRTC_RELAY_URL").unwrap_or(&"<missing>".to_string())
        );
        tracing::debug!(
            "  REDIS_URL: {}",
            env.get("REDIS_URL").unwrap_or(&"<missing>".to_string())
        );

        Some(env)
    };

    let mut pids = HashMap::new();
    let mut failures = Vec::new();
    for node_name in node_names {
        let should_use_node_runner = use_node_runner;

        tracing::debug!(
            "🚀 Spawning node: {} (use_node_runner: {})",
            node_name,
            should_use_node_runner
        );
        let result = if should_use_node_runner {
            process_service.spawn_node_runner(node_name, project_name, project_env.clone())
        } else {
            process_service.spawn_node_direct(node_name, project_name, project_env.clone())
        };

        match result {
            Ok(pid) => {
                pids.insert(node_name.clone(), pid);
                let _ = std::fs::OpenOptions::new()
                    .create(true)
                    .append(true)
                    .open("/tmp/command-listener.log")
                    .and_then(|mut f| {
                        use std::io::Write;
                        writeln!(f, "[spawn_nodes_by_name] Spawned '{}' with PID {}", node_name, pid)
                    });
                if let Some(buffer) = log_buffer {
                    buffer.push(format!("✅ STARTED: {} (PID: {})", node_name, pid));
                }
            }
            Err(e) => {
                eprintln!("Failed to spawn {}: {}", node_name, e);
                if let Some(buffer) = log_buffer {
                    buffer.push(format!("ERROR: Failed to start {}: {}", node_name, e));
                }
                failures.push((node_name.clone(), e.to_string()));
            }
        }
    }

    for (node, pid) in &pids {
        println!("  ✅ {} (PID: {})", node, pid);
    }
    println!();

    Ok(SpawnOutcome { pids, failures })
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn injects_visible_env_var_when_sim_visible_and_simulator_present() {
        let mut env = HashMap::new();
        inject_sim_visible_env(true, &["simulator".to_string(), "teleop".to_string()], &mut env);
        assert_eq!(env.get(SIMULATOR_VISIBLE_ENV_VAR), Some(&"1".to_string()));
    }

    #[test]
    fn injects_visible_env_var_for_scoped_simulator_node_name() {
        let mut env = HashMap::new();
        inject_sim_visible_env(true, &["@mecha10/simulator".to_string()], &mut env);
        assert_eq!(env.get(SIMULATOR_VISIBLE_ENV_VAR), Some(&"1".to_string()));
    }

    #[test]
    fn does_not_inject_when_sim_visible_not_requested() {
        let mut env = HashMap::new();
        inject_sim_visible_env(false, &["simulator".to_string()], &mut env);
        assert!(!env.contains_key(SIMULATOR_VISIBLE_ENV_VAR));
    }

    #[test]
    fn does_not_inject_when_no_simulator_node_in_batch() {
        let mut env = HashMap::new();
        inject_sim_visible_env(true, &["teleop".to_string(), "camera".to_string()], &mut env);
        assert!(!env.contains_key(SIMULATOR_VISIBLE_ENV_VAR));
    }

    #[test]
    fn does_not_inject_when_node_list_is_empty() {
        let mut env = HashMap::new();
        inject_sim_visible_env(true, &[], &mut env);
        assert!(!env.contains_key(SIMULATOR_VISIBLE_ENV_VAR));
    }

    fn project_config_with_simulation(
        simulation: Option<crate::types::project::ProjectSimulationConfig>,
    ) -> ProjectConfig {
        use crate::types::project::{NodesConfig, RobotConfig};

        ProjectConfig {
            name: "test".to_string(),
            version: "0.1.0".to_string(),
            template_version: None,
            robot: RobotConfig {
                id: "test".to_string(),
                name: "test".to_string(),
                platform: None,
                description: None,
            },
            simulation,
            nodes: NodesConfig::new(),
            targets: Default::default(),
            behaviors: None,
            docker: Default::default(),
            environments: Default::default(),
            actuator: Default::default(),
        }
    }

    #[test]
    fn injects_simulation_override_env_when_simulator_present() {
        use crate::types::project::ProjectSimulationConfig;

        let config = project_config_with_simulation(Some(ProjectSimulationConfig {
            scenario: None,
            model: Some("humanoid".to_string()),
            environment: Some("basic_arena".to_string()),
            preset: Some("default".to_string()),
        }));
        let mut env = HashMap::new();
        inject_simulation_override_env(&config, &["simulator".to_string(), "teleop".to_string()], &mut env);
        assert_eq!(env.get(SIMULATOR_MODEL_ENV_VAR), Some(&"humanoid".to_string()));
        assert_eq!(env.get(SIMULATOR_ENV_ENV_VAR), Some(&"basic_arena".to_string()));
        assert_eq!(env.get(SIMULATOR_PRESET_ENV_VAR), Some(&"default".to_string()));
    }

    #[test]
    fn injects_simulation_override_env_for_scoped_simulator_node_name() {
        use crate::types::project::ProjectSimulationConfig;

        let config = project_config_with_simulation(Some(ProjectSimulationConfig {
            scenario: None,
            model: Some("humanoid".to_string()),
            environment: None,
            preset: None,
        }));
        let mut env = HashMap::new();
        inject_simulation_override_env(&config, &["@mecha10/simulator".to_string()], &mut env);
        assert_eq!(env.get(SIMULATOR_MODEL_ENV_VAR), Some(&"humanoid".to_string()));
        assert!(!env.contains_key(SIMULATOR_ENV_ENV_VAR));
        assert!(!env.contains_key(SIMULATOR_PRESET_ENV_VAR));
    }

    #[test]
    fn does_not_inject_simulation_override_env_when_no_simulator_node_in_batch() {
        use crate::types::project::ProjectSimulationConfig;

        let config = project_config_with_simulation(Some(ProjectSimulationConfig {
            scenario: None,
            model: Some("humanoid".to_string()),
            environment: Some("basic_arena".to_string()),
            preset: Some("default".to_string()),
        }));
        let mut env = HashMap::new();
        inject_simulation_override_env(&config, &["teleop".to_string()], &mut env);
        assert!(env.is_empty());
    }

    #[test]
    fn injects_simulation_override_env_for_preset_only() {
        use crate::types::project::ProjectSimulationConfig;

        let config = project_config_with_simulation(Some(ProjectSimulationConfig {
            scenario: None,
            model: None,
            environment: None,
            preset: Some("cats".to_string()),
        }));
        let mut env = HashMap::new();
        inject_simulation_override_env(&config, &["simulator".to_string()], &mut env);
        assert_eq!(env.get(SIMULATOR_PRESET_ENV_VAR), Some(&"cats".to_string()));
        assert!(!env.contains_key(SIMULATOR_MODEL_ENV_VAR));
        assert!(!env.contains_key(SIMULATOR_ENV_ENV_VAR));
    }

    #[test]
    fn does_not_inject_simulation_override_env_when_project_has_no_simulation_config() {
        let config = project_config_with_simulation(None);
        let mut env = HashMap::new();
        inject_simulation_override_env(&config, &["simulator".to_string()], &mut env);
        assert!(env.is_empty());
    }

    fn project_config_with_locomotion_limits(
        locomotion: crate::types::project::LocomotionLimitsConfig,
    ) -> ProjectConfig {
        project_config_with_actuator_config(crate::types::project::ActuatorConfig {
            locomotion,
            joints: Vec::new(),
        })
    }

    fn project_config_with_actuator_config(actuator: crate::types::project::ActuatorConfig) -> ProjectConfig {
        use crate::types::project::{NodesConfig, RobotConfig};

        ProjectConfig {
            name: "test".to_string(),
            version: "0.1.0".to_string(),
            template_version: None,
            robot: RobotConfig {
                id: "test".to_string(),
                name: "test".to_string(),
                platform: None,
                description: None,
            },
            simulation: None,
            nodes: NodesConfig::new(),
            targets: Default::default(),
            behaviors: None,
            docker: Default::default(),
            environments: Default::default(),
            actuator,
        }
    }

    #[test]
    fn injects_configured_linear_and_angular_vel_regardless_of_node_batch() {
        use crate::types::project::LocomotionLimitsConfig;

        let config = project_config_with_locomotion_limits(LocomotionLimitsConfig {
            max_linear_vel: 10.0,
            max_angular_vel: 2.5,
            ..Default::default()
        });
        let mut env = HashMap::new();
        inject_locomotion_limits_env(&config, &mut env);
        assert_eq!(env.get("MECHA10_LOCOMOTION_MAX_LINEAR_SPEED"), Some(&"10".to_string()));
        assert_eq!(
            env.get("MECHA10_LOCOMOTION_MAX_ANGULAR_SPEED"),
            Some(&"2.5".to_string())
        );
    }

    #[test]
    fn injects_compiled_defaults_when_project_omits_actuator_config() {
        let config = project_config_with_locomotion_limits(Default::default());
        let mut env = HashMap::new();
        inject_locomotion_limits_env(&config, &mut env);
        assert_eq!(
            env.get("MECHA10_LOCOMOTION_MAX_LINEAR_SPEED"),
            Some(&mecha10_core::config::defaults::default_max_linear_vel().to_string())
        );
        assert_eq!(
            env.get("MECHA10_LOCOMOTION_MAX_ANGULAR_SPEED"),
            Some(&mecha10_core::config::defaults::default_max_angular_vel().to_string())
        );
        assert_eq!(
            env.get("MECHA10_LOCOMOTION_ACCELERATION"),
            Some(&mecha10_core::config::defaults::default_max_acceleration().to_string())
        );
    }

    #[test]
    fn injects_acceleration_when_configured() {
        use crate::types::project::LocomotionLimitsConfig;

        let config = project_config_with_locomotion_limits(LocomotionLimitsConfig {
            max_acceleration: 3.5,
            ..Default::default()
        });
        let mut env = HashMap::new();
        inject_locomotion_limits_env(&config, &mut env);
        assert_eq!(env.get("MECHA10_LOCOMOTION_ACCELERATION"), Some(&"3.5".to_string()));
    }

    #[test]
    fn injects_compiled_acceleration_default_when_unconfigured() {
        let config = project_config_with_locomotion_limits(Default::default());
        let mut env = HashMap::new();
        inject_locomotion_limits_env(&config, &mut env);
        assert_eq!(
            env.get("MECHA10_LOCOMOTION_ACCELERATION"),
            Some(&mecha10_core::config::defaults::default_max_acceleration().to_string())
        );
    }

    #[test]
    fn injects_empty_joint_limits_array_when_project_omits_joints() {
        let config = project_config_with_locomotion_limits(Default::default());
        let mut env = HashMap::new();
        inject_locomotion_limits_env(&config, &mut env);
        assert_eq!(
            env.get(mecha10_core::config::ACTUATOR_JOINT_LIMITS_ENV_VAR),
            Some(&"[]".to_string())
        );
    }

    #[test]
    fn injects_configured_joint_limits_as_json() {
        use crate::types::project::{ActuatorConfig, JointLimitConfig, LocomotionLimitsConfig};

        let config = project_config_with_actuator_config(ActuatorConfig {
            locomotion: LocomotionLimitsConfig::default(),
            joints: vec![JointLimitConfig {
                joint_id: "shoulder".to_string(),
                min_angle_rad: -1.5,
                max_angle_rad: 1.5,
            }],
        });
        let mut env = HashMap::new();
        inject_locomotion_limits_env(&config, &mut env);
        assert_eq!(
            env.get(mecha10_core::config::ACTUATOR_JOINT_LIMITS_ENV_VAR),
            Some(&r#"[{"joint_id":"shoulder","min_angle_rad":-1.5,"max_angle_rad":1.5}]"#.to_string())
        );
    }

    #[test]
    fn needs_turn_credentials_when_camera_node_present() {
        assert!(batch_needs_turn_credentials(&[
            "teleop".to_string(),
            "camera".to_string()
        ]));
    }

    #[test]
    fn needs_turn_credentials_for_scoped_video_streamer_node_name() {
        assert!(batch_needs_turn_credentials(&["@mecha10/video-streamer".to_string()]));
    }

    #[test]
    fn does_not_need_turn_credentials_when_no_video_relevant_node() {
        assert!(!batch_needs_turn_credentials(&[
            "teleop".to_string(),
            "motor".to_string()
        ]));
    }

    #[test]
    fn does_not_need_turn_credentials_when_batch_is_empty() {
        assert!(!batch_needs_turn_credentials(&[]));
    }

    /// Spawns a background thread that accepts one TCP connection and writes back an HTTP
    /// response with the given status line and body. Returns the base URL to point the fetch at.
    fn spawn_one_shot_http_server(status_line: &'static str, body: &'static str) -> String {
        let listener = std::net::TcpListener::bind("127.0.0.1:0").expect("bind ephemeral port");
        let addr = listener.local_addr().expect("local_addr");
        std::thread::spawn(move || {
            use std::io::{Read, Write};
            if let Ok((mut stream, _)) = listener.accept() {
                let mut buf = [0u8; 1024];
                let _ = stream.read(&mut buf);
                let response = format!(
                    "{status_line}\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{body}",
                    body.len()
                );
                let _ = stream.write_all(response.as_bytes());
                let _ = stream.flush();
            }
        });
        format!("http://{}", addr)
    }

    /// Binds an ephemeral port and immediately drops the listener, so a connection attempt to
    /// the returned base URL reliably fails (connection refused) - used to simulate mecha10-api
    /// being unreachable.
    fn unreachable_base_url() -> String {
        let listener = std::net::TcpListener::bind("127.0.0.1:0").expect("bind ephemeral port");
        let addr = listener.local_addr().expect("local_addr");
        drop(listener);
        format!("http://{}", addr)
    }

    #[tokio::test]
    #[serial_test::serial(turn_url_env_var)]
    async fn successful_fetch_populates_project_env() {
        let base_url = spawn_one_shot_http_server(
            "HTTP/1.1 200 OK",
            "{\"url\":\"turn:turn.example.com:3478\",\"username\":\"u\",\"credential\":\"c\"}",
        );
        let mut env = HashMap::new();

        maybe_inject_turn_credentials(&["camera".to_string()], &base_url, Some("test-api-key"), &mut env).await;

        assert_eq!(
            env.get(TURN_URL_ENV_VAR),
            Some(&"turn:turn.example.com:3478".to_string())
        );
        assert_eq!(env.get(TURN_USERNAME_ENV_VAR), Some(&"u".to_string()));
        assert_eq!(env.get(TURN_CREDENTIAL_ENV_VAR), Some(&"c".to_string()));
    }

    #[tokio::test]
    #[serial_test::serial(turn_url_env_var)]
    async fn not_configured_404_leaves_project_env_unchanged() {
        let base_url = spawn_one_shot_http_server("HTTP/1.1 404 Not Found", "");
        let mut env = HashMap::new();

        maybe_inject_turn_credentials(&["camera".to_string()], &base_url, Some("test-api-key"), &mut env).await;

        assert!(env.is_empty());
    }

    #[tokio::test]
    #[serial_test::serial(turn_url_env_var)]
    async fn network_failure_leaves_project_env_unchanged() {
        let base_url = unreachable_base_url();
        let mut env = HashMap::new();

        maybe_inject_turn_credentials(&["camera".to_string()], &base_url, Some("test-api-key"), &mut env).await;

        assert!(env.is_empty());
    }

    #[tokio::test]
    #[serial_test::serial(turn_url_env_var)]
    async fn skips_fetch_when_batch_has_no_video_relevant_node() {
        let base_url = unreachable_base_url();
        let mut env = HashMap::new();

        maybe_inject_turn_credentials(&["teleop".to_string()], &base_url, Some("test-api-key"), &mut env).await;

        assert!(env.is_empty());
    }

    #[tokio::test]
    #[serial_test::serial(turn_url_env_var)]
    async fn explicit_override_skips_fetch_and_project_env_is_preserved_unchanged() {
        std::env::set_var(TURN_URL_ENV_VAR, "turn:operator-set.example.com:3478");

        let base_url = unreachable_base_url();
        let mut env = HashMap::new();

        maybe_inject_turn_credentials(&["camera".to_string()], &base_url, Some("test-api-key"), &mut env).await;

        std::env::remove_var(TURN_URL_ENV_VAR);

        assert!(env.is_empty());
    }

    fn stub_baked_config() -> TurnConfig {
        TurnConfig {
            url: "turn:baked.example.com:3478".to_string(),
            username: "baked-user".to_string(),
            credential: "baked-secret".to_string(),
        }
    }

    #[tokio::test]
    #[serial_test::serial(turn_url_env_var)]
    async fn tier2_baked_default_populates_project_env_when_no_explicit_override() {
        let base_url = unreachable_base_url();
        let mut env = HashMap::new();

        maybe_inject_turn_credentials_with_baked_resolver(
            &["camera".to_string()],
            &base_url,
            Some("test-api-key"),
            &mut env,
            || Some(stub_baked_config()),
        )
        .await;

        assert_eq!(
            env.get(TURN_URL_ENV_VAR),
            Some(&"turn:baked.example.com:3478".to_string())
        );
        assert_eq!(env.get(TURN_USERNAME_ENV_VAR), Some(&"baked-user".to_string()));
        assert_eq!(env.get(TURN_CREDENTIAL_ENV_VAR), Some(&"baked-secret".to_string()));
    }

    #[tokio::test]
    #[serial_test::serial(turn_url_env_var)]
    async fn tier1_explicit_env_var_wins_over_tier2_baked_default() {
        std::env::set_var(TURN_URL_ENV_VAR, "turn:operator-set.example.com:3478");

        let base_url = unreachable_base_url();
        let mut env = HashMap::new();

        maybe_inject_turn_credentials_with_baked_resolver(
            &["camera".to_string()],
            &base_url,
            Some("test-api-key"),
            &mut env,
            || Some(stub_baked_config()),
        )
        .await;

        std::env::remove_var(TURN_URL_ENV_VAR);

        assert!(env.is_empty());
    }

    #[tokio::test]
    #[serial_test::serial(turn_url_env_var)]
    async fn tier3_fetch_still_runs_when_no_explicit_override_and_no_baked_default() {
        let base_url = spawn_one_shot_http_server(
            "HTTP/1.1 200 OK",
            "{\"url\":\"turn:fetched.example.com:3478\",\"username\":\"fetched-user\",\"credential\":\"fetched-secret\"}",
        );
        let mut env = HashMap::new();

        maybe_inject_turn_credentials_with_baked_resolver(
            &["camera".to_string()],
            &base_url,
            Some("test-api-key"),
            &mut env,
            || None,
        )
        .await;

        assert_eq!(
            env.get(TURN_URL_ENV_VAR),
            Some(&"turn:fetched.example.com:3478".to_string())
        );
        assert_eq!(env.get(TURN_USERNAME_ENV_VAR), Some(&"fetched-user".to_string()));
        assert_eq!(env.get(TURN_CREDENTIAL_ENV_VAR), Some(&"fetched-secret".to_string()));
    }
}