torzy 0.1.0

//! Minimal Tor-routed worker launching.
//!
//! `torzy` starts one Tor SOCKS route per worker, prepares an isolated worker
//! workspace, and runs a caller-supplied shell command once per worker. The CLI
//! and library both call [`run`], so embedded and command-line behavior stay on
//! the same code path.
//!
//! The default command is intentionally local and quiet. Network checks are left
//! to the caller-supplied command.
//!
//! ```no_run
//! use torzy::{run, LaunchConfig};
//!
//! #[tokio::main]
//! async fn main() -> anyhow::Result<()> {
//!     let summary = run(
//!         LaunchConfig {
//!             envs: 5,
//!             command: "printf 'worker=%s ready\\n' \"$TOR_WORKER_ID\"".to_string(),
//!             ..LaunchConfig::default()
//!         },
//!         None,
//!     )
//!     .await?;
//!
//!     println!("{} workers succeeded", summary.succeeded);
//!     Ok(())
//! }
//! ```

use std::collections::{BTreeMap, HashSet};
use std::ffi::OsString;
use std::net::{Ipv4Addr, Ipv6Addr};
use std::path::{Path, PathBuf};
use std::process::Stdio;
use std::sync::Arc;
use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH};

use anyhow::{Context, Result, bail};
use serde::Serialize;
#[cfg(unix)]
use std::os::unix::process::CommandExt;
use tokio::io::{AsyncBufReadExt, BufReader};
use tokio::process::{Child, Command};
use tokio::sync::{Semaphore, mpsc};
use tokio::time::{sleep, timeout};

/// Local default command used when no command is supplied.
pub const DEFAULT_COMMAND: &str = "printf 'torzy worker %s ready\\n' \"$TOR_WORKER_ID\"";

/// Configuration for one `torzy` run.
#[derive(Clone, Debug)]
pub struct LaunchConfig {
    /// Number of workers and Tor routes to prepare. Values below `1` become `1`.
    pub envs: usize,
    /// Shell command run once per worker through `/bin/sh -lc`.
    pub command: String,
    /// Explicit SOCKS routes, one per worker. Leave empty to auto-generate routes.
    pub tor_routes: Vec<String>,
    /// First SOCKS port used for auto-generated local routes.
    pub base_socks_port: u16,
    /// Per-worker timeout.
    pub timeout: Duration,
    /// Maximum workers running at once. Defaults to available CPU parallelism.
    pub max_concurrent: Option<usize>,
    /// Per-worker memory limit in MiB. `0` means unlimited.
    pub memory_limit_mb: u64,
    /// Root directory for runtime data. Defaults to a temporary directory.
    pub root_dir: Option<PathBuf>,
    /// Keep worker workspaces after completion.
    pub keep_workspaces: bool,
    /// Start and stop Tor instances automatically.
    pub manage_tor: bool,
    /// Tor executable path.
    pub tor_binary: String,
    /// Per-instance Tor bootstrap timeout.
    pub tor_timeout: Duration,
    /// Persistent Tor data directory. Defaults to runtime data under `root_dir`.
    pub tor_data_dir: Option<PathBuf>,
    /// Do not start Tor or run real commands; emit simulated success events.
    pub dry_run: bool,
    /// Export proxy environment variables to each worker.
    pub proxy_env: bool,
    /// Command working directory. Defaults to each isolated worker workspace.
    pub working_directory: Option<PathBuf>,
    /// Extra environment variables for workers.
    pub extra_env: BTreeMap<String, String>,
}

impl Default for LaunchConfig {
    fn default() -> Self {
        Self {
            envs: 5,
            command: DEFAULT_COMMAND.to_string(),
            tor_routes: Vec::new(),
            base_socks_port: 9050,
            timeout: Duration::from_secs(120),
            max_concurrent: None,
            memory_limit_mb: 0,
            root_dir: None,
            keep_workspaces: false,
            manage_tor: true,
            tor_binary: "tor".to_string(),
            tor_timeout: Duration::from_secs(90),
            tor_data_dir: None,
            dry_run: false,
            proxy_env: true,
            working_directory: None,
            extra_env: BTreeMap::new(),
        }
    }
}

/// Output stream that produced a worker line.
#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize)]
pub enum StreamKind {
    /// Worker standard output.
    Stdout,
    /// Worker standard error.
    Stderr,
}

impl StreamKind {
    /// Stable lowercase label.
    pub fn as_str(self) -> &'static str {
        match self {
            Self::Stdout => "stdout",
            Self::Stderr => "stderr",
        }
    }
}

/// Final worker status.
#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize)]
pub enum WorkerStatus {
    /// Worker exited successfully.
    Succeeded,
    /// Worker exited unsuccessfully or could not spawn.
    Failed,
    /// Worker exceeded its timeout.
    TimedOut,
    /// Worker was terminated by the launcher.
    Killed,
}

impl WorkerStatus {
    /// Stable lowercase label.
    pub fn as_str(self) -> &'static str {
        match self {
            Self::Succeeded => "succeeded",
            Self::Failed => "failed",
            Self::TimedOut => "timed_out",
            Self::Killed => "killed",
        }
    }
}

/// Runtime event emitted by [`run`].
#[derive(Clone, Debug)]
pub enum TorzyEvent {
    /// Tor process was started for a worker.
    TorStarted { id: usize },
    /// Tor bootstrap percentage advanced.
    TorBootstrap { id: usize, pct: u8 },
    /// Tor route is ready.
    TorReady { id: usize },
    /// Tor route failed to start or bootstrap.
    TorFailed { id: usize, detail: String },
    /// Managed Tor process was stopped.
    TorStopped { id: usize },
    /// All routes are ready and workers are launching.
    WorkersReady,
    /// Worker process was started.
    WorkerStarted { id: usize },
    /// Redacted worker output line.
    WorkerOutput {
        id: usize,
        stream: StreamKind,
        line: String,
    },
    /// Worker reached a terminal status.
    WorkerFinished {
        id: usize,
        status: WorkerStatus,
        exit_code: Option<i32>,
        detail: Option<String>,
    },
}

/// JSON-friendly worker summary.
#[derive(Clone, Debug, Serialize)]
pub struct WorkerSummary {
    /// Zero-based worker id.
    pub id: usize,
    /// Final status label.
    pub status: String,
    /// Process exit code when available.
    pub exit_code: Option<i32>,
    /// Redacted failure or timeout detail when available.
    pub detail: Option<String>,
    /// Worker runtime in milliseconds.
    pub duration_ms: u128,
}

/// JSON-friendly run summary that avoids command, route, path, and output data.
#[derive(Clone, Debug, Serialize)]
pub struct RunSummary {
    /// Start time as seconds since Unix epoch.
    pub started_at_unix: u64,
    /// End time as seconds since Unix epoch.
    pub ended_at_unix: u64,
    /// Number of configured workers.
    pub total_envs: usize,
    /// Maximum worker concurrency used for the run.
    pub max_concurrent: usize,
    /// Per-worker memory limit in MiB.
    pub memory_limit_mb: u64,
    /// Whether `torzy` managed Tor processes.
    pub managed_tor: bool,
    /// Whether this was a dry run.
    pub dry_run: bool,
    /// Whether proxy environment variables were exported to workers.
    pub proxy_env: bool,
    /// Workers that reached a terminal status.
    pub completed: usize,
    /// Workers that succeeded.
    pub succeeded: usize,
    /// Workers that failed.
    pub failed: usize,
    /// Workers that timed out.
    pub timed_out: usize,
    /// Workers killed by the launcher.
    pub killed: usize,
    /// Per-worker status summaries.
    pub workers: Vec<WorkerSummary>,
}

#[derive(Clone)]
struct WorkerConfig {
    id: usize,
    command: String,
    tor_route: String,
    timeout: Duration,
    root: PathBuf,
    workspace: PathBuf,
    working_directory: Option<PathBuf>,
    memory_limit_mb: u64,
    keep_workspace: bool,
    proxy_env: bool,
    extra_env: BTreeMap<String, String>,
}

struct ManagedTor {
    id: usize,
    child: Child,
}

/// Run the configured Tor-routed workers.
///
/// The optional event sender receives sanitized progress and worker output. The
/// returned summary intentionally omits commands, routes, filesystem paths, and
/// output lines.
pub async fn run(
    config: LaunchConfig,
    events: Option<mpsc::UnboundedSender<TorzyEvent>>,
) -> Result<RunSummary> {
    let envs = config.envs.max(1);
    let max_concurrent = config
        .max_concurrent
        .unwrap_or_else(|| {
            std::thread::available_parallelism()
                .map(|n| n.get())
                .unwrap_or(4)
        })
        .max(1);
    let tor_routes = routes_for_config(&config, envs)?;

    let started = Instant::now();
    let started_at_unix = unix_now();
    let root = config.root_dir.clone().unwrap_or_else(|| {
        std::env::temp_dir().join(format!("torzy-{started_at_unix}-{}", std::process::id()))
    });
    tokio::fs::create_dir_all(&root)
        .await
        .context("creating root directory")?;

    let mut tor_instances = if config.manage_tor && !config.dry_run {
        start_tor_instances(&config, &tor_routes, &root, &events).await?
    } else {
        Vec::new()
    };

    emit(&events, TorzyEvent::WorkersReady);

    let semaphore = Arc::new(Semaphore::new(max_concurrent));
    let mut handles = Vec::with_capacity(envs);
    for (id, tor_route) in tor_routes.iter().enumerate() {
        let workspace = root.join("workers").join(format!("worker-{id}"));
        let worker = WorkerConfig {
            id,
            command: config.command.clone(),
            tor_route: tor_route.clone(),
            timeout: config.timeout,
            root: root.clone(),
            workspace,
            working_directory: config.working_directory.clone(),
            memory_limit_mb: config.memory_limit_mb,
            keep_workspace: config.keep_workspaces,
            proxy_env: config.proxy_env,
            extra_env: config.extra_env.clone(),
        };
        let events = events.clone();
        let semaphore = semaphore.clone();
        let dry_run = config.dry_run;
        handles.push((
            id,
            tokio::spawn(async move { run_worker(worker, dry_run, semaphore, events).await }),
        ));
    }

    let mut workers = Vec::with_capacity(envs);
    for (id, handle) in handles {
        match handle.await {
            Ok(summary) => workers.push(summary),
            Err(error) => workers.push(WorkerSummary {
                id,
                status: WorkerStatus::Failed.as_str().to_string(),
                exit_code: None,
                detail: Some(redact_detail(&format!("worker task join error: {error}"))),
                duration_ms: 0,
            }),
        }
    }
    workers.sort_by_key(|worker| worker.id);

    stop_tor_instances(&mut tor_instances, &events).await;
    if !config.keep_workspaces && config.tor_data_dir.is_none() {
        let _ = tokio::fs::remove_dir_all(root.join("tor-data")).await;
    }

    let ended_at_unix = unix_now();
    Ok(build_summary(
        started_at_unix,
        ended_at_unix,
        &config,
        max_concurrent,
        workers,
        started,
    ))
}

/// Generate local SOCKS route strings from a base port.
pub fn generate_tor_routes(env_count: usize, base_socks_port: u16) -> Result<Vec<String>> {
    let mut routes = Vec::with_capacity(env_count.max(1));
    for id in 0..env_count.max(1) {
        let offset = u16::try_from(id).context("too many environments for SOCKS port range")?;
        let port = base_socks_port
            .checked_add(offset)
            .context("base SOCKS port plus environment count exceeds port range")?;
        routes.push(format!("socks5h://127.0.0.1:{port}"));
    }
    Ok(routes)
}

/// Validate that explicit routes are non-empty, unique, and match worker count.
pub fn validate_tor_routes(routes: &[String], env_count: usize) -> Result<()> {
    if routes.len() != env_count.max(1) {
        bail!(
            "got {} tor route(s), but env count is {}; provide exactly one route per worker",
            routes.len(),
            env_count.max(1),
        );
    }

    let mut unique = HashSet::new();
    for route in routes {
        if route.trim().is_empty() {
            bail!("tor routes cannot be empty");
        }
        if !unique.insert(route) {
            bail!("duplicate tor route provided");
        }
    }

    Ok(())
}

/// Expand command placeholders for a worker.
pub fn expand_command_template(
    raw: &str,
    tor_route: &str,
    worker_id: usize,
    root: &Path,
    workspace: &Path,
) -> String {
    raw.replace("{TOR_ROUTE}", tor_route)
        .replace("{TOR_WORKER_ID}", &worker_id.to_string())
        .replace("{WORKER_ID}", &worker_id.to_string())
        .replace("{TORZY_ROOT}", root.to_string_lossy().as_ref())
        .replace("{TORZY_WORKSPACE}", workspace.to_string_lossy().as_ref())
}

/// Redact route strings, runtime paths, and public network-address tokens.
pub fn redact_line(line: &str) -> String {
    redact_line_with_context(line, None, None)
}

fn routes_for_config(config: &LaunchConfig, envs: usize) -> Result<Vec<String>> {
    if config.tor_routes.is_empty() {
        generate_tor_routes(envs, config.base_socks_port)
    } else {
        validate_tor_routes(&config.tor_routes, envs)?;
        Ok(config.tor_routes.clone())
    }
}

async fn start_tor_instances(
    config: &LaunchConfig,
    routes: &[String],
    root: &Path,
    events: &Option<mpsc::UnboundedSender<TorzyEvent>>,
) -> Result<Vec<ManagedTor>> {
    let tor_data_root = config
        .tor_data_dir
        .clone()
        .unwrap_or_else(|| root.join("tor-data"));
    tokio::fs::create_dir_all(&tor_data_root)
        .await
        .context("creating tor data directory")?;

    let mut instances = Vec::with_capacity(routes.len());
    let mut boot_handles = Vec::with_capacity(routes.len());

    for (id, route) in routes.iter().enumerate() {
        let port = extract_port(route)?;
        let data_dir = tor_data_root.join(format!("tor-{port}"));
        tokio::fs::create_dir_all(&data_dir)
            .await
            .context("creating tor data directory")?;

        let empty_torrc = data_dir.join("torrc");
        tokio::fs::write(&empty_torrc, "")
            .await
            .context("writing empty torrc")?;

        let mut child = Command::new(&config.tor_binary)
            .arg("-f")
            .arg(&empty_torrc)
            .arg("--SocksPort")
            .arg(port.to_string())
            .arg("--DataDirectory")
            .arg(&data_dir)
            .arg("--PidFile")
            .arg(data_dir.join("tor.pid"))
            .arg("--Log")
            .arg("notice stderr")
            .stdout(Stdio::null())
            .stderr(Stdio::piped())
            .stdin(Stdio::null())
            .spawn()
            .with_context(|| "failed to start tor binary; install tor or pass --tor-binary")?;

        emit(events, TorzyEvent::TorStarted { id });
        let stderr = child
            .stderr
            .take()
            .context("managed tor process did not expose stderr")?;
        let events_for_reader = events.clone();
        boot_handles.push((
            id,
            tokio::spawn(async move { read_tor_bootstrap(id, stderr, events_for_reader).await }),
        ));
        instances.push(ManagedTor { id, child });
    }

    for (id, handle) in boot_handles {
        match timeout(config.tor_timeout, handle).await {
            Ok(Ok(Ok(()))) => {}
            Ok(Ok(Err(error))) => {
                let detail = redact_detail(&error.to_string());
                emit(&events, TorzyEvent::TorFailed { id, detail });
                stop_tor_instances(&mut instances, events).await;
                return Err(error);
            }
            Ok(Err(error)) => {
                let detail = redact_detail(&format!("tor bootstrap task failed: {error}"));
                emit(
                    &events,
                    TorzyEvent::TorFailed {
                        id,
                        detail: detail.clone(),
                    },
                );
                stop_tor_instances(&mut instances, events).await;
                bail!(detail);
            }
            Err(_) => {
                let detail = format!(
                    "tor did not bootstrap within {}s",
                    config.tor_timeout.as_secs()
                );
                emit(
                    &events,
                    TorzyEvent::TorFailed {
                        id,
                        detail: detail.clone(),
                    },
                );
                stop_tor_instances(&mut instances, events).await;
                bail!(detail);
            }
        }
    }

    Ok(instances)
}

async fn read_tor_bootstrap<R>(
    id: usize,
    source: R,
    events: Option<mpsc::UnboundedSender<TorzyEvent>>,
) -> Result<()>
where
    R: tokio::io::AsyncRead + Unpin,
{
    let reader = BufReader::new(source);
    let mut lines = reader.lines();
    let mut last_pct = 0;

    while let Some(line) = lines.next_line().await? {
        if let Some(pct) = parse_bootstrap_pct(&line) {
            if pct > last_pct {
                last_pct = pct;
                emit(&events, TorzyEvent::TorBootstrap { id, pct });
            }
            if pct >= 100 {
                emit(&events, TorzyEvent::TorReady { id });
                return Ok(());
            }
        }
    }

    bail!("tor exited before bootstrap completed")
}

async fn stop_tor_instances(
    instances: &mut [ManagedTor],
    events: &Option<mpsc::UnboundedSender<TorzyEvent>>,
) {
    for instance in instances {
        let _ = instance.child.kill().await;
        let _ = instance.child.wait().await;
        emit(events, TorzyEvent::TorStopped { id: instance.id });
    }
}

async fn run_worker(
    config: WorkerConfig,
    dry_run: bool,
    semaphore: Arc<Semaphore>,
    events: Option<mpsc::UnboundedSender<TorzyEvent>>,
) -> WorkerSummary {
    let _permit = semaphore.acquire().await.expect("semaphore closed");
    let started = Instant::now();

    if dry_run {
        emit(&events, TorzyEvent::WorkerStarted { id: config.id });
        for i in 0..3 {
            emit(
                &events,
                TorzyEvent::WorkerOutput {
                    id: config.id,
                    stream: StreamKind::Stdout,
                    line: format!("dry-run {i}: worker ready"),
                },
            );
            sleep(Duration::from_millis(40)).await;
        }
        emit(
            &events,
            TorzyEvent::WorkerFinished {
                id: config.id,
                status: WorkerStatus::Succeeded,
                exit_code: Some(0),
                detail: Some("dry run done".to_string()),
            },
        );
        return worker_summary(
            &config,
            WorkerStatus::Succeeded,
            Some(0),
            Some("dry run done".to_string()),
            started,
        );
    }

    if let Err(error) = prepare_worker_workspace(&config).await {
        let detail = redact_detail(&error.to_string());
        emit(
            &events,
            TorzyEvent::WorkerFinished {
                id: config.id,
                status: WorkerStatus::Failed,
                exit_code: None,
                detail: Some(detail.clone()),
            },
        );
        return worker_summary(&config, WorkerStatus::Failed, None, Some(detail), started);
    }

    let command = expand_command_template(
        &config.command,
        &config.tor_route,
        config.id,
        &config.root,
        &config.workspace,
    );
    let current_dir = config
        .working_directory
        .clone()
        .unwrap_or_else(|| config.workspace.clone());

    let cargo_home = config.workspace.join(".cargo");
    let cargo_target_dir = config.workspace.join("target");
    let worker_path = worker_path(&cargo_home);
    let mem_bytes = if config.memory_limit_mb > 0 {
        config.memory_limit_mb.saturating_mul(1024 * 1024)
    } else {
        0
    };

    let mut std_cmd = std::process::Command::new("/bin/sh");
    std_cmd
        .arg("-lc")
        .arg(&command)
        .current_dir(&current_dir)
        .env("HOME", &config.workspace)
        .env("TMPDIR", config.workspace.join(".tmp"))
        .env("CARGO_HOME", &cargo_home)
        .env("CARGO_TARGET_DIR", &cargo_target_dir)
        .env("PATH", worker_path)
        .env("TOR_WORKER_ID", config.id.to_string())
        .env("TOR_ROUTE", &config.tor_route)
        .env("TORZY_ROOT", &config.root)
        .env("TORZY_WORKSPACE", &config.workspace)
        .stdout(Stdio::piped())
        .stderr(Stdio::piped())
        .stdin(Stdio::null());

    if config.proxy_env {
        for key in [
            "ALL_PROXY",
            "HTTPS_PROXY",
            "HTTP_PROXY",
            "all_proxy",
            "https_proxy",
            "http_proxy",
            "CARGO_HTTP_PROXY",
        ] {
            std_cmd.env(key, &config.tor_route);
        }
    }

    for (key, value) in &config.extra_env {
        std_cmd.env(key, value);
    }

    #[cfg(unix)]
    unsafe {
        std_cmd.pre_exec(move || {
            libc::setsid();
            if mem_bytes > 0 {
                let lim = libc::rlimit {
                    rlim_cur: mem_bytes as libc::rlim_t,
                    rlim_max: mem_bytes as libc::rlim_t,
                };
                libc::setrlimit(libc::RLIMIT_AS, &lim);
                libc::setrlimit(libc::RLIMIT_RSS, &lim);
            }
            libc::nice(19);
            Ok(())
        });
    }

    let mut child = match Command::from(std_cmd).spawn() {
        Ok(child) => child,
        Err(error) => {
            let detail = redact_detail(&format!("failed to spawn worker: {error}"));
            emit(
                &events,
                TorzyEvent::WorkerFinished {
                    id: config.id,
                    status: WorkerStatus::Failed,
                    exit_code: None,
                    detail: Some(detail.clone()),
                },
            );
            return worker_summary(&config, WorkerStatus::Failed, None, Some(detail), started);
        }
    };

    emit(&events, TorzyEvent::WorkerStarted { id: config.id });

    if let Some(stdout) = child.stdout.take() {
        tokio::spawn(capture_lines(
            config.id,
            StreamKind::Stdout,
            stdout,
            config.root.clone(),
            config.workspace.clone(),
            events.clone(),
        ));
    }
    if let Some(stderr) = child.stderr.take() {
        tokio::spawn(capture_lines(
            config.id,
            StreamKind::Stderr,
            stderr,
            config.root.clone(),
            config.workspace.clone(),
            events.clone(),
        ));
    }

    let (wait_result, timed_out) = tokio::select! {
        result = child.wait() => (result, false),
        _ = sleep(config.timeout) => {
            kill_worker(&mut child).await;
            (child.wait().await, true)
        }
    };

    let (status, exit_code, detail) = match wait_result {
        Ok(status) if timed_out => (
            WorkerStatus::TimedOut,
            status.code(),
            Some(format!("timed out at {}s", config.timeout.as_secs())),
        ),
        Ok(status) if status.success() => (WorkerStatus::Succeeded, status.code(), None),
        Ok(status) => (
            WorkerStatus::Failed,
            status.code(),
            Some(format!("exit code {}", status.code().unwrap_or(-1))),
        ),
        Err(error) => (
            WorkerStatus::Failed,
            None,
            Some(redact_detail(&format!("worker wait error: {error}"))),
        ),
    };

    emit(
        &events,
        TorzyEvent::WorkerFinished {
            id: config.id,
            status,
            exit_code,
            detail: detail.clone(),
        },
    );

    if !config.keep_workspace {
        let _ = tokio::fs::remove_dir_all(&config.workspace).await;
    }

    worker_summary(&config, status, exit_code, detail, started)
}

async fn prepare_worker_workspace(config: &WorkerConfig) -> Result<()> {
    for path in [
        config.workspace.clone(),
        config.workspace.join(".tmp"),
        config.workspace.join(".cargo"),
        config.workspace.join("target"),
    ] {
        tokio::fs::create_dir_all(&path)
            .await
            .context("creating worker directory")?;
    }
    Ok(())
}

async fn capture_lines<R>(
    id: usize,
    stream: StreamKind,
    source: R,
    root: PathBuf,
    workspace: PathBuf,
    events: Option<mpsc::UnboundedSender<TorzyEvent>>,
) where
    R: tokio::io::AsyncRead + Unpin,
{
    let reader = BufReader::new(source);
    let mut lines = reader.lines();

    while let Ok(Some(line)) = lines.next_line().await {
        let line = redact_line_with_context(&line, Some(&root), Some(&workspace));
        emit(&events, TorzyEvent::WorkerOutput { id, stream, line });
    }
}

async fn kill_worker(child: &mut Child) {
    #[cfg(unix)]
    if let Some(pid) = child.id() {
        let pgid = -(pid as i32);
        unsafe {
            libc::kill(pgid, libc::SIGTERM);
        }
        sleep(Duration::from_secs(3)).await;
        unsafe {
            libc::kill(pgid, libc::SIGKILL);
        }
    }

    let _ = child.kill().await;
}

fn worker_path(cargo_home: &Path) -> OsString {
    let mut entries = vec![cargo_home.join("bin")];
    if let Some(path) = std::env::var_os("PATH") {
        entries.extend(std::env::split_paths(&path));
    }
    std::env::join_paths(entries).unwrap_or_else(|_| OsString::new())
}

fn worker_summary(
    config: &WorkerConfig,
    status: WorkerStatus,
    exit_code: Option<i32>,
    detail: Option<String>,
    started: Instant,
) -> WorkerSummary {
    WorkerSummary {
        id: config.id,
        status: status.as_str().to_string(),
        exit_code,
        detail: detail.map(|value| redact_detail(&value)),
        duration_ms: started.elapsed().as_millis(),
    }
}

fn build_summary(
    started_at_unix: u64,
    ended_at_unix: u64,
    config: &LaunchConfig,
    max_concurrent: usize,
    workers: Vec<WorkerSummary>,
    _started: Instant,
) -> RunSummary {
    let mut succeeded = 0;
    let mut failed = 0;
    let mut timed_out = 0;
    let mut killed = 0;

    for worker in &workers {
        match worker.status.as_str() {
            "succeeded" => succeeded += 1,
            "timed_out" => timed_out += 1,
            "killed" => killed += 1,
            _ => failed += 1,
        }
    }

    RunSummary {
        started_at_unix,
        ended_at_unix,
        total_envs: workers.len(),
        max_concurrent,
        memory_limit_mb: config.memory_limit_mb,
        managed_tor: config.manage_tor,
        dry_run: config.dry_run,
        proxy_env: config.proxy_env,
        completed: workers.len(),
        succeeded,
        failed,
        timed_out,
        killed,
        workers,
    }
}

fn extract_port(route: &str) -> Result<u16> {
    route
        .rsplit(':')
        .next()
        .and_then(|value| value.parse::<u16>().ok())
        .with_context(|| "route does not end with a valid port")
}

fn parse_bootstrap_pct(line: &str) -> Option<u8> {
    let start = line.find("Bootstrapped ")? + "Bootstrapped ".len();
    let pct_end = line[start..].find('%')? + start;
    line[start..pct_end].trim().parse::<u8>().ok()
}

fn redact_detail(value: &str) -> String {
    redact_line(value)
}

fn redact_line_with_context(line: &str, root: Option<&Path>, workspace: Option<&Path>) -> String {
    let mut value = line.to_string();

    if let Some(workspace) = workspace {
        let workspace = workspace.to_string_lossy();
        if !workspace.is_empty() {
            value = value.replace(workspace.as_ref(), "[torzy-workspace]");
        }
    }
    if let Some(root) = root {
        let root = root.to_string_lossy();
        if !root.is_empty() {
            value = value.replace(root.as_ref(), "[torzy-root]");
        }
    }

    value
        .split_whitespace()
        .map(redact_token)
        .collect::<Vec<_>>()
        .join(" ")
}

fn redact_token(token: &str) -> String {
    let trimmed = token.trim_matches(|ch: char| {
        matches!(
            ch,
            ',' | ';' | ')' | '(' | '[' | ']' | '{' | '}' | '<' | '>' | '"' | '\''
        )
    });

    if trimmed.starts_with("socks5://") || trimmed.starts_with("socks5h://") {
        return token.replace(trimmed, "[tor-route]");
    }

    if should_redact_addr(trimmed) {
        return token.replace(trimmed, "[addr-redacted]");
    }

    token.to_string()
}

fn should_redact_addr(value: &str) -> bool {
    if let Ok(addr) = value.parse::<Ipv4Addr>() {
        return redact_v4(addr);
    }
    if let Ok(addr) = value.parse::<Ipv6Addr>() {
        return !addr.is_loopback() && !addr.is_unspecified();
    }
    false
}

fn redact_v4(addr: Ipv4Addr) -> bool {
    let octets = addr.octets();
    !(addr.is_loopback()
        || addr.is_private()
        || addr.is_link_local()
        || addr.is_unspecified()
        || addr.is_broadcast()
        || octets[0] >= 224)
}

fn unix_now() -> u64 {
    SystemTime::now()
        .duration_since(UNIX_EPOCH)
        .unwrap_or_default()
        .as_secs()
}

fn emit(events: &Option<mpsc::UnboundedSender<TorzyEvent>>, event: TorzyEvent) {
    if let Some(events) = events {
        let _ = events.send(event);
    }
}

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

    #[test]
    fn generates_routes_from_base_port() {
        assert_eq!(
            generate_tor_routes(3, 9100).unwrap(),
            vec![
                "socks5h://127.0.0.1:9100",
                "socks5h://127.0.0.1:9101",
                "socks5h://127.0.0.1:9102",
            ]
        );
    }

    #[test]
    fn rejects_duplicate_routes() {
        let routes = vec![
            "socks5h://127.0.0.1:9050".to_string(),
            "socks5h://127.0.0.1:9050".to_string(),
        ];
        assert!(validate_tor_routes(&routes, 2).is_err());
    }

    #[test]
    fn expands_command_template() {
        let command = expand_command_template(
            "route={TOR_ROUTE} id={TOR_WORKER_ID} root={TORZY_ROOT} work={TORZY_WORKSPACE}",
            "socks5h://127.0.0.1:9050",
            7,
            Path::new("/tmp/root"),
            Path::new("/tmp/root/workers/worker-7"),
        );
        assert_eq!(
            command,
            "route=socks5h://127.0.0.1:9050 id=7 root=/tmp/root work=/tmp/root/workers/worker-7"
        );
    }

    #[test]
    fn parses_tor_bootstrap_percentage() {
        assert_eq!(
            parse_bootstrap_pct("Bootstrapped 100% (done): Done"),
            Some(100)
        );
        assert_eq!(parse_bootstrap_pct("not bootstrap"), None);
    }

    #[test]
    fn redacts_sensitive_tokens() {
        assert_eq!(
            redact_line("route socks5h://127.0.0.1:9050 public 203.0.113.9 local 127.0.0.1"),
            "route [tor-route] public [addr-redacted] local 127.0.0.1"
        );
    }

    #[tokio::test]
    async fn dry_run_succeeds_without_tor() {
        let root = std::env::temp_dir().join(format!("torzy-test-{}", unix_now()));
        let summary = run(
            LaunchConfig {
                envs: 2,
                dry_run: true,
                manage_tor: false,
                root_dir: Some(root.clone()),
                max_concurrent: Some(2),
                ..LaunchConfig::default()
            },
            None,
        )
        .await
        .unwrap();

        assert_eq!(summary.total_envs, 2);
        assert_eq!(summary.succeeded, 2);
        assert_eq!(summary.workers[0].status, "succeeded");
        let _ = tokio::fs::remove_dir_all(root).await;
    }
}