cargowatch_detector/

lib.rs

//! Best-effort external Rust process detection.

use std::collections::{BTreeMap, BTreeSet};
use std::path::{Path, PathBuf};

use time::OffsetDateTime;
use tokio::sync::mpsc;
use tokio::time::{Duration, sleep};
use tracing::debug;

use cargowatch_core::{DetectedProcess, DetectedProcessClass, SessionEvent, detected_session_id};

/// Detects Rust-related processes using `sysinfo`.
pub struct RustProcessDetector {
    system: sysinfo::System,
}

impl Default for RustProcessDetector {
    fn default() -> Self {
        let system = sysinfo::System::new_all();
        Self { system }
    }
}

impl RustProcessDetector {
    /// Perform a point-in-time scan for active Rust-related processes.
    pub fn scan(&mut self) -> Vec<DetectedProcess> {
        self.system
            .refresh_processes(sysinfo::ProcessesToUpdate::All, true);
        let now = OffsetDateTime::now_utc();
        let mut processes = Vec::new();

        for (pid, process) in self.system.processes() {
            let process_name = process.name().to_string_lossy().to_string();
            let command = process
                .cmd()
                .iter()
                .map(|part| part.to_string_lossy().to_string())
                .collect::<Vec<_>>();
            if !is_rust_process(&process_name, &command) {
                continue;
            }

            let started_at = unix_timestamp_to_utc(process.start_time() as i64);
            let cwd = process.cwd().map(|path| path.to_path_buf());
            let workspace_root = infer_workspace_root(cwd.as_deref(), &command);
            let classification = classify_process(&process_name, &command);
            let session_id = detected_session_id(pid.as_u32(), started_at);
            let elapsed_ms = i64::try_from((now - started_at).whole_milliseconds())
                .unwrap_or(i64::MAX)
                .max(0);

            processes.push(DetectedProcess {
                session_id,
                pid: pid.as_u32(),
                process_name,
                command,
                cwd,
                workspace_root,
                classification,
                started_at,
                last_seen_at: now,
                elapsed_ms,
            });
        }

        processes.sort_by(|left, right| right.started_at.cmp(&left.started_at));
        processes
    }
}

/// Background poller that emits point-in-time process change events.
pub struct DetectionService {
    detector: RustProcessDetector,
    sender: mpsc::UnboundedSender<SessionEvent>,
    poll_interval: Duration,
    previous: BTreeMap<String, DetectedProcess>,
}

impl DetectionService {
    /// Create a new detection service.
    pub fn new(sender: mpsc::UnboundedSender<SessionEvent>, poll_interval_ms: u64) -> Self {
        Self {
            detector: RustProcessDetector::default(),
            sender,
            poll_interval: Duration::from_millis(poll_interval_ms),
            previous: BTreeMap::new(),
        }
    }

    /// Run the detection loop until the cancellation token fires.
    pub async fn run(mut self, cancellation: tokio_util::sync::CancellationToken) {
        loop {
            self.tick();
            tokio::select! {
                _ = cancellation.cancelled() => return,
                _ = sleep(self.poll_interval) => {}
            }
        }
    }

    fn tick(&mut self) {
        let current = self
            .detector
            .scan()
            .into_iter()
            .map(|process| (process.session_id.clone(), process))
            .collect::<BTreeMap<_, _>>();

        for (session_id, process) in &current {
            match self.previous.get(session_id) {
                None => {
                    let _ = self
                        .sender
                        .send(SessionEvent::ProcessDetected(process.clone()));
                }
                Some(previous) if previous != process => {
                    let _ = self
                        .sender
                        .send(SessionEvent::ProcessUpdated(process.clone()));
                }
                _ => {}
            }
        }

        let current_ids = current.keys().cloned().collect::<BTreeSet<_>>();
        for (session_id, previous) in &self.previous {
            if !current_ids.contains(session_id) {
                let _ = self.sender.send(SessionEvent::ProcessGone {
                    session_id: previous.session_id.clone(),
                    pid: previous.pid,
                    observed_at: OffsetDateTime::now_utc(),
                });
            }
        }

        debug!(count = current.len(), "detector tick completed");
        self.previous = current;
    }
}

fn is_rust_process(process_name: &str, command: &[String]) -> bool {
    let first = command
        .first()
        .map(|part| executable_stem(part))
        .unwrap_or_else(|| executable_stem(process_name));
    matches!(
        first.as_deref(),
        Some("cargo" | "rustc" | "rustdoc" | "clippy-driver")
    )
}

fn classify_process(process_name: &str, command: &[String]) -> DetectedProcessClass {
    let stem = command
        .first()
        .map(|part| executable_stem(part))
        .unwrap_or_else(|| executable_stem(process_name))
        .unwrap_or_default();
    match stem.as_str() {
        "cargo" => {
            let subcommand = command.iter().skip(1).find(|part| !part.starts_with('-'));
            match subcommand.map(String::as_str) {
                Some("build") => DetectedProcessClass::CargoBuild,
                Some("check") => DetectedProcessClass::CargoCheck,
                Some("test") => DetectedProcessClass::CargoTest,
                Some("clippy") => DetectedProcessClass::CargoClippy,
                Some("doc") => DetectedProcessClass::CargoDoc,
                _ => DetectedProcessClass::UnknownRustProcess,
            }
        }
        "rustc" => DetectedProcessClass::RustcCompile,
        "rustdoc" => DetectedProcessClass::Rustdoc,
        "clippy-driver" => DetectedProcessClass::CargoClippy,
        _ => DetectedProcessClass::UnknownRustProcess,
    }
}

fn infer_workspace_root(cwd: Option<&Path>, command: &[String]) -> Option<PathBuf> {
    if let Some(manifest_path) = manifest_path_from_command(cwd, command) {
        let manifest_dir = if manifest_path.ends_with("Cargo.toml") {
            manifest_path.parent().map(Path::to_path_buf)
        } else {
            Some(manifest_path)
        };
        if let Some(root) = manifest_dir {
            return find_workspace_root(&root);
        }
    }
    cwd.and_then(find_workspace_root)
}

fn manifest_path_from_command(cwd: Option<&Path>, command: &[String]) -> Option<PathBuf> {
    let manifest = command
        .windows(2)
        .find(|pair| pair[0] == "--manifest-path")
        .map(|pair| PathBuf::from(&pair[1]))?;
    if manifest.is_absolute() {
        Some(manifest)
    } else {
        cwd.map(|cwd| cwd.join(manifest))
    }
}

fn find_workspace_root(start: &Path) -> Option<PathBuf> {
    let mut last_match = None;
    for ancestor in start.ancestors() {
        if ancestor.join("Cargo.toml").exists() {
            last_match = Some(ancestor.to_path_buf());
        }
    }
    last_match
}

fn executable_stem(value: &str) -> Option<String> {
    Path::new(value)
        .file_stem()
        .and_then(|stem| stem.to_str())
        .map(|stem| stem.to_string())
}

fn unix_timestamp_to_utc(timestamp: i64) -> OffsetDateTime {
    OffsetDateTime::from_unix_timestamp(timestamp).unwrap_or(OffsetDateTime::UNIX_EPOCH)
}

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

    #[test]
    fn classifies_cargo_subcommands() {
        let class = classify_process("cargo", &["cargo".into(), "check".into()]);
        assert_eq!(class, DetectedProcessClass::CargoCheck);
    }

    #[test]
    fn infers_workspace_root_from_manifest_argument() {
        let root = tempfile::tempdir().expect("tempdir");
        let member = root.path().join("member");
        std::fs::create_dir_all(&member).expect("member dir");
        std::fs::write(
            root.path().join("Cargo.toml"),
            "[workspace]\nmembers=[\"member\"]",
        )
        .expect("workspace manifest");
        std::fs::write(
            member.join("Cargo.toml"),
            "[package]\nname=\"member\"\nversion=\"0.1.0\"\nedition=\"2024\"",
        )
        .expect("member manifest");

        let detected = infer_workspace_root(
            None,
            &[
                "cargo".into(),
                "check".into(),
                "--manifest-path".into(),
                member.join("Cargo.toml").display().to_string(),
            ],
        );

        assert_eq!(detected, Some(root.path().to_path_buf()));
    }

    #[test]
    fn resolves_relative_manifest_paths_against_process_cwd() {
        let process_root = tempfile::tempdir().expect("tempdir");
        let workspace = process_root.path().join("workspace");
        let member = workspace.join("member");
        std::fs::create_dir_all(&member).expect("member dir");
        std::fs::write(
            workspace.join("Cargo.toml"),
            "[workspace]\nmembers=[\"member\"]",
        )
        .expect("workspace manifest");
        std::fs::write(
            member.join("Cargo.toml"),
            "[package]\nname=\"member\"\nversion=\"0.1.0\"\nedition=\"2024\"",
        )
        .expect("member manifest");

        let detected = infer_workspace_root(
            Some(process_root.path()),
            &[
                "cargo".into(),
                "check".into(),
                "--manifest-path".into(),
                "workspace/member/Cargo.toml".into(),
            ],
        );

        assert_eq!(detected, Some(workspace));
    }
}