net-mumu 0.2.0-rc.3

// src/manager.rs
//
// NetManager — spawns and coordinates `fetch` tasks for the MuMu
// dynamic plugin. Tasks run on background threads and post `NetMessage`s into
// an internal channel. The REPL’s poller (installed in src/bridge.rs) calls
// `poll_events(...)`, which drains the channel and invokes user-provided
// callbacks **on the interpreter thread**.
//
// This module focuses solely on `net:fetch` (ping is implemented as a direct
// transform in bridge.rs and does not go through the manager).

use std::collections::HashMap;
use std::process::{Command, Stdio};
use std::sync::mpsc::{self, Receiver, Sender};
use std::sync::{
    atomic::{AtomicUsize, Ordering},
    LockResult, Mutex, MutexGuard, OnceLock,
};
use std::thread;

use mumu::{FunctionValue, Interpreter, Value};
use mumu::parser::core::driver::parse_tokens;
use mumu::parser::lexer::tokenize;

use crate::bridge::NetMessage;

/* ────────────────────────────────────────────────────────────────────────── */
/* Public globals                                                             */
/* ────────────────────────────────────────────────────────────────────────── */

/// Tracks how many background tasks are active (fetch combined).
/// The REPL poller consults this to decide whether more polling is needed.
pub static ACTIVE_TASKS: AtomicUsize = AtomicUsize::new(0);

/// A lazy global that provides a `lock()` method so callers can do
/// `NET_MANAGER.lock().unwrap()` conveniently.
pub struct NetManagerGlobal(OnceLock<Mutex<NetManager>>);

pub static NET_MANAGER: NetManagerGlobal = NetManagerGlobal(OnceLock::new());

impl NetManagerGlobal {
    /// Borrow the global manager. The guard lives no longer than `&self`.
    pub fn lock<'a>(&'a self) -> LockResult<MutexGuard<'a, NetManager>> {
        let m = self.0.get_or_init(|| Mutex::new(NetManager::new()));
        m.lock()
    }
}

/* ────────────────────────────────────────────────────────────────────────── */
/* Task records                                                               */
/* ────────────────────────────────────────────────────────────────────────── */

/// Simple record for a running `fetch` task.
struct FetchTask {
    callback: Box<FunctionValue>,
}

/* ────────────────────────────────────────────────────────────────────────── */
/* NetManager                                                                 */
/* ────────────────────────────────────────────────────────────────────────── */

/// Public so bridge.rs can call `NetManager::global()`.
pub struct NetManager {
    next_token: usize,
    tx: Sender<NetMessage>,
    rx: Receiver<NetMessage>,
    fetches: HashMap<usize, FetchTask>,
}

impl NetManager {
    pub fn new() -> Self {
        let (tx, rx) = mpsc::channel::<NetMessage>();
        Self {
            next_token: 1,
            tx,
            rx,
            fetches: HashMap::new(),
        }
    }

    /// Convenience accessor used by bridge helpers (`NetManager::global().lock()`).
    pub fn global() -> &'static Mutex<NetManager> {
        NET_MANAGER.0.get_or_init(|| Mutex::new(NetManager::new()))
    }

    /* ───────────────────────────── fetch ───────────────────────────── */

    /// Start a background HTTP fetch using `curl`. Returns a token id.
    pub fn add_fetch_task(
        &mut self,
        url: String,
        callback: Box<FunctionValue>,
        verbose: bool,
    ) -> usize {
        let token = self.alloc_token();
        self.fetches.insert(token, FetchTask { callback });

        let tx = self.tx.clone();
        ACTIVE_TASKS.fetch_add(1, Ordering::SeqCst);

        thread::spawn(move || {
            if verbose {
                eprintln!("[net:fetch] token={} url={}", token, url);
            }

            // Use `curl` to keep dependencies minimal.
            // Flags:
            //   -f  fail on HTTP errors
            //   -s  silent
            //   -S  show errors
            //   -L  follow redirects
            let mut cmd = Command::new("curl");
            cmd.arg("-fsSL")
                .arg(&url)
                .stdout(Stdio::piped())
                .stderr(Stdio::piped());

            let output = cmd.output();

            match output {
                Ok(out) if out.status.success() => {
                    let body = String::from_utf8_lossy(&out.stdout).to_string();
                    let _ = tx.send(NetMessage::FetchOk(token, body));
                }
                Ok(out) => {
                    let err = String::from_utf8_lossy(&out.stderr).to_string();
                    let msg = if err.trim().is_empty() {
                        format!("curl exited with {}", out.status)
                    } else {
                        err
                    };
                    let _ = tx.send(NetMessage::FetchErr(token, msg));
                }
                Err(e) => {
                    let _ = tx.send(NetMessage::FetchErr(
                        token,
                        format!("failed to spawn curl: {}", e),
                    ));
                }
            }
        });

        token
    }

    /* ─────────────────────────── event pump ─────────────────────────── */

    /// Drain pending messages and invoke user callbacks on the interpreter thread.
    pub fn poll_events(&mut self, interp: &mut Interpreter) {
        while let Ok(msg) = self.rx.try_recv() {
            match msg {
                /* ---------------------- fetch ---------------------- */
                NetMessage::FetchOk(token, body) => {
                    if let Some(task) = self.fetches.remove(&token) {
                        let _ = Self::call_callback(interp, &task.callback, Value::SingleString(body));
                    }
                    ACTIVE_TASKS.fetch_sub(1, Ordering::SeqCst);
                }
                NetMessage::FetchErr(token, err) => {
                    if let Some(task) = self.fetches.remove(&token) {
                        let _ = Self::call_callback(
                            interp,
                            &task.callback,
                            Value::SingleString(format!("ERROR: {}", err)),
                        );
                    }
                    ACTIVE_TASKS.fetch_sub(1, Ordering::SeqCst);
                }

                /* ---------------------- ping (unused here) --------- */
                NetMessage::PingLine(_, _) => {
                    // No-op: ping is handled as a transform in bridge.rs now.
                }
                NetMessage::PingErr(_, _) => {
                    // No-op.
                }
                NetMessage::PingDone(_) => {
                    // No-op.
                }
            }
        }
    }

    /* ────────────────────────── internals ─────────────────────────── */

    fn alloc_token(&mut self) -> usize {
        let t = self.next_token;
        self.next_token = self.next_token.wrapping_add(1);
        if self.next_token == 0 {
            // avoid 0 token; extremely unlikely wrap
            self.next_token = 1;
        }
        t
    }

    #[inline]
    fn call_callback(
        interp: &mut Interpreter,
        cb: &Box<FunctionValue>,
        arg: Value,
    ) -> Result<(), String> {
        // Generic, parser-driven invocation that works for any FunctionValue.
        // We avoid depending on internal apply helpers from MuMu core.
        interp.set_variable("__net_cb", Value::Function(cb.clone()));
        interp.set_variable("__net_arg", arg);

        let code = "__net_cb(__net_arg)";
        let tokens = tokenize(code, interp.is_verbose())
            .map_err(|e| format!("callback tokenize error: {}", e))?;
        let ast = parse_tokens(&tokens, interp.is_verbose())
            .map_err(|e| format!("callback parse error: {}", e))?;
        for stmt in ast.iter() {
            let _ = interp.exec_statement(stmt)?;
        }
        Ok(())
    }
}