mumu-process 0.1.0

// process/src/info.rs
//
// Manages "process:info" tasks. Now increments/decrements a global ACTIVE_TASKS
// so the test harness can see how many tasks remain and waits for them.

use mumu::{
    Interpreter,
    Value,
    FunctionValue,
    apply_one_function_value,
};
use std::{
    env,
    process,
    sync::{
        Arc,
        Mutex,
        mpsc::{channel, Sender, Receiver},
    },
    thread,
    collections::HashMap,
};
use lazy_static::lazy_static;
use sysinfo::{System, SystemExt, ProcessExt, Pid, PidExt};
use whoami;
use indexmap::IndexMap;

// Import the shared counter from lib.rs
use crate::ACTIVE_TASKS;
use std::sync::atomic::Ordering;

// On Unix, we can get the actual UID; on other OS, fallback to 0
#[cfg(unix)]
fn get_uid() -> i32 {
    use nix::unistd::getuid;
    getuid().as_raw() as i32
}

#[cfg(not(unix))]
fn get_uid() -> i32 {
    0
}

enum ProcessMessage {
    Ok(usize, Value),
    Err(usize, String),
}

struct ProcessTask {
    callback: Box<FunctionValue>,
    done: bool,
}

struct ProcessManager {
    next_id: usize,
    tasks: HashMap<usize, ProcessTask>,
    tx: Sender<ProcessMessage>,
    rx: Receiver<ProcessMessage>,
}

lazy_static! {
    static ref PROCESS_MANAGER: Mutex<ProcessManager> = Mutex::new(ProcessManager::new());
}

impl ProcessManager {
    fn new() -> Self {
        let (tx, rx) = channel();
        ProcessManager {
            next_id: 0,
            tasks: HashMap::new(),
            tx,
            rx,
        }
    }

    /// Creates a new background task for 'process:info'
    /// and increments ACTIVE_TASKS so the harness sees it.
    fn add_process_info_task(&mut self, callback: Box<FunctionValue>, verbose: bool) -> usize {
        let task_id = self.next_id;
        self.next_id += 1;

        if verbose {
            eprintln!("[process plugin] add_process_info_task => Creating new task with id={}.", task_id);
        }

        self.tasks.insert(task_id, ProcessTask {
            callback,
            done: false,
        });

        // Just like net plugin, increment ACTIVE_TASKS
        ACTIVE_TASKS.fetch_add(1, Ordering::SeqCst);

        if verbose {
            let total = ACTIVE_TASKS.load(Ordering::SeqCst);
            eprintln!("[process plugin] add_process_info_task => ACTIVE_TASKS => now total={}.", total);
        }

        let txc = self.tx.clone();
        thread::spawn(move || {
            if verbose {
                eprintln!("[process plugin] (thread) => Starting background gather for task_id={}.", task_id);
            }

            // 1) Create a sysinfo::System
            let mut sys = System::new();
            sys.refresh_processes();

            // 2) Attempt to read memory usage in KB
            let pid_u32 = process::id();
            let pid_sys: Pid = Pid::from_u32(pid_u32);
            let mem_kb = match sys.process(pid_sys) {
                Some(proc_info) => {
                    if verbose {
                        eprintln!(
                            "[process plugin] (thread) => Found process info in sysinfo. Memory = {} KB",
                            proc_info.memory()
                        );
                    }
                    proc_info.memory()
                }
                None => {
                    if verbose {
                        eprintln!("[process plugin] (thread) => No process info for pid={}, using 0KB", pid_u32);
                    }
                    0
                }
            };

            let uid_val = get_uid();
            let username_val = whoami::username();
            let exe_full = match env::current_exe() {
                Ok(p) => p,
                Err(_e) => std::path::PathBuf::from("<unknown>"),
            };
            let self_str = exe_full.display().to_string();
            let binary_name = exe_full
                .file_name()
                .map(|os| os.to_string_lossy().to_string())
                .unwrap_or_else(|| "<no_binary_name>".to_string());
            let cwd_str = match env::current_dir() {
                Ok(p) => p.display().to_string(),
                Err(_e) => "<unknown>".to_string(),
            };
            let launch_args: Vec<String> = env::args().collect();
            let platform_str = env::consts::OS;
            let arch_str = env::consts::ARCH;
            // just an example
            let event_loop_len = 42;

            // Build the keyed array
            let mut map = IndexMap::new();
            map.insert("pid".to_string(), Value::Int(pid_u32 as i32));
            map.insert("uid".to_string(), Value::Int(uid_val));
            map.insert("username".to_string(), Value::SingleString(username_val));
            map.insert("self".to_string(), Value::SingleString(self_str));
            map.insert("binary_name".to_string(), Value::SingleString(binary_name));
            map.insert("cwd".to_string(), Value::SingleString(cwd_str));
            map.insert("launch_args".to_string(), Value::StrArray(launch_args));
            map.insert("platform".to_string(), Value::SingleString(platform_str.to_string()));
            map.insert("architecture".to_string(), Value::SingleString(arch_str.to_string()));
            map.insert("memory_usage_kb".to_string(), Value::Int(mem_kb as i32));
            map.insert("event_loop_len".to_string(), Value::Int(event_loop_len));

            let result_val = Value::KeyedArray(map);

            if verbose {
                eprintln!(
                    "[process plugin] (thread) => Gathering done => sending Ok(task_id={}).",
                    task_id
                );
            }

            if txc.send(ProcessMessage::Ok(task_id, result_val)).is_err() {
                eprintln!("[process plugin] (thread) => Could not send Ok(...) for task_id={}", task_id);
            }
        });

        task_id
    }

    /// poll_events => receives messages, calls callbacks, removes done tasks, updates ACTIVE_TASKS
    fn poll_events(&mut self, interp: &mut Interpreter) {
        let verbose = interp.is_verbose();
        if verbose {
            eprintln!("[process plugin] poll_events => about to check channel for messages...");
        }

        while let Ok(msg) = self.rx.try_recv() {
            if verbose {
                eprintln!("[process plugin] poll_events => got a message => {:?}", describe_message(&msg));
            }
            match msg {
                ProcessMessage::Ok(id, data) => {
                    if let Some(task) = self.tasks.get_mut(&id) {
                        if !task.done {
                            task.done = true;
                            if verbose {
                                eprintln!("[process plugin] poll_events => calling user callback => task_id={}.", id);
                            }
                            let _ = apply_one_function_value(interp, task.callback.clone(), data);
                        }
                    }
                }
                ProcessMessage::Err(id, errmsg) => {
                    if let Some(task) = self.tasks.get_mut(&id) {
                        if !task.done {
                            task.done = true;
                            if verbose {
                                eprintln!(
                                    "[process plugin] poll_events => callback with error => task_id={}, '{}'",
                                    id, errmsg
                                );
                            }
                            let val = Value::StrArray(vec![errmsg]);
                            let _ = apply_one_function_value(interp, task.callback.clone(), val);
                        }
                    }
                }
            }
        }

        // remove tasks that are done
        let before = self.tasks.len();
        self.tasks.retain(|_, t| !t.done);
        let after = self.tasks.len();
        let removed = before.saturating_sub(after);
        if removed > 0 {
            // decrement global ACTIVE_TASKS
            ACTIVE_TASKS.fetch_sub(removed, Ordering::SeqCst);
            if verbose {
                eprintln!(
                    "[process plugin] poll_events => removed {} tasks, ACTIVE_TASKS now={}",
                    removed,
                    ACTIVE_TASKS.load(Ordering::SeqCst)
                );
            }
        }
    }

    /// return the current ACTIVE_TASKS count
    fn count_active_tasks(&self) -> usize {
        ACTIVE_TASKS.load(Ordering::SeqCst)
    }
}

fn describe_message(msg: &ProcessMessage) -> String {
    match msg {
        ProcessMessage::Ok(id, _) => format!("Ok(task_id={})", id),
        ProcessMessage::Err(id, e) => format!("Err(task_id={}, '{}')", id, e),
    }
}