net-mumu 0.2.0-rc.3

// LOCAL/src/bridge.rs
// src/bridge.rs
//
// Dynamic bridge for the MuMu "net" plugin.
//
// Exposes:
//   - net:fetch(url, cb)
//   - net:ping(options|string)         -> 0-arg transform that emits one row per call
//   - net:real_ping(options|string)    -> raw-ICMP iterator adapted to a transform
//   - net:lldp(options|string|[...])   -> LLDP/CDP capture as a 0-arg transform
//
// Notes:
//   • LLDP/CDP engine is used by default.
//   • If the caller sets `stub:true` in options (or the module is configured
//     to force the stub via environment in lldp::mod.rs), the synthetic stub
//     stream is used instead.
//   • When the engine cannot run (e.g., missing CAP_NET_RAW), it will emit a
//     single `ok:false` row and then return "AGAIN" (non-blocking) on pulls,
//     so Flow pipelines stay alive and can throttle/stop appropriately.
//
// (c) 2025 — MIT/Apache-2.0

use std::ffi::c_void;
use std::process::{Command, Stdio};
use std::sync::{Arc, Mutex};

use indexmap::IndexMap;
use mumu::{FunctionValue, Interpreter, Value};

use crate::manager::{NetManager, ACTIVE_TASKS, NET_MANAGER};
use crate::util::resolve_host;

pub enum NetMessage {
    PingLine(usize, String),
    PingDone(usize),
    PingErr(usize, String),
    FetchOk(usize, String),
    FetchErr(usize, String),
}

/* ───────────────────────────── net:fetch(url, callback) ───────────────────────────── */

pub fn fetch_bridge_fn(interp: &mut Interpreter, mut args: Vec<Value>) -> Result<Value, String> {
    if args.len() != 2 {
        return Err(format!(
            "net:fetch(url, callback) => expected 2 arguments, got {}",
            args.len()
        ));
    }

    let url_str = match args.remove(0) {
        Value::SingleString(s) => s,
        Value::StrArray(ss) if ss.len() == 1 => ss[0].clone(),
        _ => return Err("net:fetch => first arg must be a single string".to_string()),
    };

    let cb_func = match args.remove(0) {
        Value::Function(fb) => fb,
        other => return Err(format!("net:fetch => second arg must be function, got {:?}", other)),
    };

    Ok(handle_fetch_call(interp, &url_str, cb_func))
}

pub fn handle_fetch_call(interp: &mut Interpreter, url: &str, cb: Box<FunctionValue>) -> Value {
    let mut mgr = NetManager::global().lock().unwrap();
    let _token_id = mgr.add_fetch_task(url.to_string(), cb, interp.is_verbose());
    Value::Bool(true)
}

/* ───────────────────────────── helpers ───────────────────────────── */

fn get_string_like(v: &Value) -> Option<String> {
    match v {
        Value::SingleString(s) => Some(s.clone()),
        Value::StrArray(ss) if ss.len() == 1 => Some(ss[0].clone()),
        _ => None,
    }
}

fn strings_from(v: &Value) -> Option<Vec<String>> {
    match v {
        Value::SingleString(s) => Some(vec![s.clone()]),
        Value::StrArray(ss) => Some(ss.clone()),
        _ => None,
    }
}

fn to_u64_ms(v: &Value) -> Option<u64> {
    match v {
        Value::Int(i) if *i >= 0 => Some(*i as u64),
        Value::Long(l) if *l >= 0 => Some(*l as u64),
        Value::Float(f) if *f >= 0.0 => Some(*f as u64),
        Value::SingleString(s) => s.trim().parse::<f64>().ok().filter(|x| *x >= 0.0).map(|x| x as u64),
        _ => None,
    }
}
fn to_u32(v: &Value) -> Option<u32> {
    match v {
        Value::Int(i) if *i >= 0 => Some(*i as u32),
        Value::Long(l) if *l >= 0 => Some(*l as u32),
        Value::Float(f) if *f >= 0.0 => Some(*f as u32),
        Value::SingleString(s) => s
            .trim()
            .parse::<i64>()
            .ok()
            .filter(|x| *x >= 0)
            .map(|x| x as u32),
        _ => None,
    }
}
fn to_u16(v: &Value) -> Option<u16> {
    match v {
        Value::Int(i) if *i >= 0 && *i <= u16::MAX as i32 => Some(*i as u16),
        Value::Long(l) if *l >= 0 && *l <= u16::MAX as i64 => Some(*l as u16),
        Value::Float(f) if *f >= 0.0 && *f <= u16::MAX as f64 => Some(*f as u16),
        Value::SingleString(s) => s.trim().parse::<u64>().ok().filter(|x| *x <= u16::MAX as u64).map(|x| x as u16),
        _ => None,
    }
}
fn to_usize(v: &Value) -> Option<usize> {
    match v {
        Value::Int(i) if *i >= 0 => Some(*i as usize),
        Value::Long(l) if *l >= 0 => Some(*l as usize),
        Value::Float(f) if *f >= 0.0 => Some((*f as u64) as usize),
        Value::SingleString(s) => s.trim().parse::<u64>().ok().map(|x| x as usize),
        _ => None,
    }
}
fn to_bool(v: &Value) -> Option<bool> {
    match v {
        Value::Bool(b) => Some(*b),
        Value::Int(i) => Some(*i != 0),
        Value::Long(l) => Some(*l != 0),
        Value::SingleString(s) => {
            let t = s.trim().to_ascii_lowercase();
            Some(matches!(t.as_str(), "1" | "true" | "yes" | "on"))
        }
        _ => None,
    }
}

/* ───────────────────────────── net:ping(options) -> transform ──────────────────────────
   (system ping; performs one probe per pull)
*/
pub fn ping_bridge_fn(_interp: &mut Interpreter, mut args: Vec<Value>) -> Result<Value, String> {
    if args.len() != 1 {
        return Err(format!(
            "net:ping(options) => expected 1 argument (dest string or keyed options), got {}",
            args.len()
        ));
    }

    // Parse options
    let dest = match args.remove(0) {
        Value::SingleString(s) => s,
        Value::StrArray(ss) if ss.len() == 1 => ss[0].clone(),
        Value::KeyedArray(map) => {
            let v = map
                .get("dest")
                .or_else(|| map.get("host"))
                .ok_or_else(|| "net:ping => options must include 'dest' (or 'host')".to_string())?;
            get_string_like(v).ok_or_else(|| "net:ping => 'dest'/'host' must be a string".to_string())?
        }
        _ => return Err("net:ping => first arg must be string or keyed options".to_string()),
    };

    // Resolve once for the display/ip field (best-effort).
    let ip_addr = resolve_host(&dest).unwrap_or_else(|_| "0.0.0.0".parse().unwrap());
    let ip_s = ip_addr.to_string();

    // Per-call state
    #[derive(Debug)]
    struct PingOneShotEnv {
        dest: String,
        ip: String,
        seq: i32,
        timeout_ms: u64,
    }

    let env = Arc::new(Mutex::new(PingOneShotEnv {
        dest,
        ip: ip_s,
        seq: 0,
        timeout_ms: 1000,
    }));

    // Transform: each invocation performs one probe and returns a KeyedArray result.
    let tf = Value::Function(Box::new(FunctionValue::RustClosure(
        "net:ping-transform".to_string(),
        Arc::new(Mutex::new(move |_interp: &mut Interpreter, _args: Vec<Value>| {
            let mut st = env
                .lock()
                .map_err(|_| "net:ping => state lock error".to_string())?;
            st.seq = st.seq.saturating_add(1);

            // Build platform-specific **one-shot** ping command
            #[cfg(unix)]
            let mut cmd = {
                let mut c = Command::new("ping");
                c.arg("-n");
                c.arg("-c").arg("1");
                let secs = (st.timeout_ms as f64 / 1000.0).max(1.0);
                c.arg("-W").arg(format!("{}", secs as i32));
                c.arg(&st.dest);
                c.stdout(Stdio::piped());
                c.stderr(Stdio::piped());
                c
            };

            #[cfg(not(unix))]
            let mut cmd = {
                let mut c = Command::new("ping");
                c.arg("-n").arg("1");
                c.arg("-w").arg(format!("{}", st.timeout_ms));
                c.arg(&st.dest);
                c.stdout(Stdio::piped());
                c.stderr(Stdio::piped());
                c
            };

            let output = cmd.output();

            // Prepare result map (no printing, no raw lines)
            let mut map: IndexMap<String, Value> = IndexMap::new();
            map.insert("dest".into(), Value::SingleString(st.dest.clone()));
            map.insert("ip".into(), Value::SingleString(st.ip.clone()));
            map.insert("seq".into(), Value::Int(st.seq));
            map.insert("ok".into(), Value::Bool(false));

            match output {
                Ok(out) => {
                    let stdout = String::from_utf8_lossy(&out.stdout).to_string();
                    let stderr = String::from_utf8_lossy(&out.stderr).to_string();
                    let lc = stdout.to_ascii_lowercase();

                    // Reply present?
                    let reply_like = lc.contains(" bytes from ")
                        || lc.contains("reply from")
                        || lc.contains("time=");

                    // Extract fields (best-effort)
                    let bytes = if let Some(p) = lc.find(" bytes from ") {
                        let left = &stdout[..p];
                        left.trim().parse::<i32>().ok()
                    } else if let Some(p) = lc.find("bytes=") {
                        let s = &stdout[p + 6..];
                        let end = s.find(|c: char| !c.is_ascii_digit()).unwrap_or(s.len());
                        s[..end].parse::<i32>().ok()
                    } else {
                        None
                    };

                    let ttl = if let Some(p) = lc.find("ttl=") {
                        let s = &stdout[p + 4..];
                        let end = s.find(|c: char| !c.is_ascii_digit()).unwrap_or(s.len());
                        s[..end].trim().parse::<i32>().ok()
                    } else {
                        None
                    };

                    let ms = if let Some(p) = lc.find("time=") {
                        let s = &stdout[p + 5..];
                        let end = s
                            .find(|c: char| !(c.is_ascii_digit() || c == '.'))
                            .unwrap_or(s.len());
                        s[..end].trim().parse::<f64>().ok()
                    } else {
                        None
                    };

                    if let Some(b) = bytes {
                        map.insert("bytes".into(), Value::Int(b));
                    }
                    if let Some(tl) = ttl {
                        map.insert("ttl".into(), Value::Int(tl));
                    }
                    if let Some(rt) = ms {
                        map.insert("ms".into(), Value::Float(rt));
                    }

                    let ok = out.status.success() && reply_like;
                    map.insert("ok".into(), Value::Bool(ok));

                    if !ok {
                        let mut msg = stderr.trim().to_string();
                        if msg.is_empty() {
                            msg = stdout
                                .lines()
                                .find(|l| {
                                    let t = l.trim();
                                    !t.is_empty()
                                        && !t.starts_with("PING ")
                                        && !t.starts_with("--- ")
                                        && !t.contains("statistics")
                                })
                                .unwrap_or("")
                                .to_string();
                        }
                        if !msg.is_empty() {
                            map.insert("message".into(), Value::SingleString(msg));
                        }
                    }

                    Ok(Value::KeyedArray(map))
                }
                Err(e) => {
                    map.insert(
                        "message".into(),
                        Value::SingleString(format!("spawn error: {}", e)),
                    );
                    Ok(Value::KeyedArray(map))
                }
            }
        })),
        0,
    )));

    Ok(tf)
}

/* ───────────────────────────── net:real_ping(options|string) ───────────────────────────── */

pub fn real_ping_bridge_fn(interp: &mut Interpreter, mut args: Vec<Value>) -> Result<Value, String> {
    if args.len() != 1 {
        return Err(format!(
            "net:real_ping(options) => expected 1 argument (dest string or keyed options), got {}",
            args.len()
        ));
    }

    // Defaults
    let mut dest: Option<String> = None;
    let mut timeout_ms: u64 = 1000;
    let mut interval_ms: u64 = 1000;
    let mut count_opt: Option<usize> = None; // None => infinite
    let mut seq_start: u16 = 1;

    match args.remove(0) {
        Value::SingleString(s) => {
            dest = Some(s);
        }
        Value::StrArray(ss) if ss.len() == 1 => {
            dest = Some(ss[0].clone());
        }
        Value::KeyedArray(map) => {
            // dest / host
            if let Some(v) = map.get("dest").or_else(|| map.get("host")) {
                dest = get_string_like(v);
            }
            // timeout_ms (accepted; engine is pull-driven anyway)
            if let Some(v) = map.get("timeout_ms").or_else(|| map.get("timeout")) {
                if let Some(ms) = to_u64_ms(v) {
                    timeout_ms = ms;
                }
            }
            // interval_ms (accepted; engine is pull-driven anyway)
            if let Some(v) = map.get("interval_ms").or_else(|| map.get("interval")) {
                if let Some(ms) = to_u64_ms(v) {
                    interval_ms = ms;
                }
            }
            // count
            if let Some(v) = map.get("count") {
                count_opt = to_usize(v);
            }
            // seq
            if let Some(v) = map.get("seq").or_else(|| map.get("sequence")) {
                if let Some(s) = to_u16(v) {
                    seq_start = s;
                }
            }
        }
        _ => return Err("net:real_ping => first arg must be string or keyed options".to_string()),
    }

    let dest = dest.ok_or_else(|| "net:real_ping => 'dest' (or 'host') is required".to_string())?;

    // Build the internal iterator, then adapt to a 0-arg **transform** for API parity.
    let handle = crate::real_ping::spawn_iterator(crate::real_ping::RealPingOptions {
        dest,
        count: count_opt,
        interval_ms: Some(interval_ms),
        timeout_ms,
        seq_start,
        verbose: interp.is_verbose(),
    });

    let tf = crate::real_ping::iter_to_transform(handle);
    Ok(Value::Function(tf))
}

/* ───────────────────────────── net:lldp(options) -> transform ────────────────────────── */

pub fn lldp_bridge_fn(interp: &mut Interpreter, mut args: Vec<Value>) -> Result<Value, String> {
    use crate::lldp::{iter_to_transform, spawn_iterator};
    use crate::lldp::{LldpMode, LldpOptions};
    use crate::lldp::proto::DiscoveryProtocol;

    if args.len() != 1 {
        return Err(format!(
            "net:lldp(options) => expected 1 argument (iface string, string array, or keyed options), got {}",
            args.len()
        ));
    }

    // Defaults
    let mut iface: Option<String> = None;
    let mut ifaces: Vec<String> = Vec::new();
    let mut mode: LldpMode = LldpMode::Listen;
    let mut ttl: u16 = 120;
    let mut count_opt: Option<usize> = None;
    let mut hostname: Option<String> = None;
    let mut port_id: Option<String> = None;
    // If true, caller explicitly requests **stub** behaviour; default is engine.
    let mut stub: bool = false;
    let mut proto_strs: Vec<String> = Vec::new();

    // Advanced capture/engine knobs
    let mut snaplen: u32 = 1518;
    let mut promisc: bool = true;
    let mut capture_timeout_ms: u32 = 100;
    let mut channel_capacity: usize = 1024;
    let mut tx_interval_ms: Option<u64> = None;
    let mut verbose_flag: Option<bool> = None; // allow explicit override

    match args.remove(0) {
        // Simple string → single interface name
        Value::SingleString(s) => {
            iface = Some(s.clone());
            ifaces.push(s);
        }
        // Array of strings → interfaces
        Value::StrArray(ss) if !ss.is_empty() => {
            ifaces = ss.clone();
            iface = ifaces.first().cloned();
        }
        // Keyed options
        Value::KeyedArray(map) => {
            // iface / ifaces
            if let Some(v) = map.get("iface")
                .or_else(|| map.get("ifname"))
                .or_else(|| map.get("interface"))
            {
                iface = get_string_like(v);
            }
            if let Some(v) = map.get("ifaces")
                .or_else(|| map.get("interfaces"))
            {
                if let Some(list) = strings_from(v) {
                    ifaces.extend(list);
                }
            }

            // mode
            if let Some(v) = map.get("mode") {
                if let Some(s) = get_string_like(v) {
                    match s.to_ascii_lowercase().as_str() {
                        "listen" => mode = LldpMode::Listen,
                        "advertise" | "adv" => mode = LldpMode::Advertise,
                        "discover" | "disc" => mode = LldpMode::Discover,
                        _ => {}
                    }
                }
            }

            // protocols (string or array of strings; commas allowed)
            if let Some(v) = map.get("protocols").or_else(|| map.get("proto")).or_else(|| map.get("protocol")) {
                if let Some(list) = strings_from(v) {
                    proto_strs.extend(list.into_iter().flat_map(|s| {
                        s.split(',')
                            .map(|t| t.trim().to_string())
                            .filter(|t| !t.is_empty())
                            .collect::<Vec<_>>()
                    }));
                }
            }

            // ttl / count / hostname / port_id / stub / verbose
            if let Some(v) = map.get("ttl") {
                if let Some(t) = to_u16(v) { ttl = t; }
            }
            if let Some(v) = map.get("count") {
                count_opt = to_usize(v);
            }
            if let Some(v) = map.get("hostname").or_else(|| map.get("sys_name")).or_else(|| map.get("system_name")) {
                hostname = get_string_like(v);
            }
            if let Some(v) = map.get("port_id").or_else(|| map.get("port")) {
                port_id = get_string_like(v);
            }
            if let Some(v) = map.get("stub") {
                if let Some(b) = to_bool(v) { stub = b; }
            }
            if let Some(v) = map.get("verbose") {
                verbose_flag = to_bool(v);
            }

            // Advanced capture knobs
            if let Some(v) = map.get("snaplen") {
                if let Some(s) = to_u32(v) { snaplen = s.max(64); }
            }
            if let Some(v) = map.get("promisc") {
                if let Some(b) = to_bool(v) { promisc = b; }
            }
            if let Some(v) = map.get("capture_timeout_ms").or_else(|| map.get("timeout_ms")) {
                if let Some(ms) = to_u32(v) { capture_timeout_ms = ms; }
            }
            if let Some(v) = map.get("channel_capacity").or_else(|| map.get("chan_cap")) {
                if let Some(n) = to_usize(v) { channel_capacity = n.max(16); }
            }
            if let Some(v) = map.get("tx_interval_ms").or_else(|| map.get("tx_interval")) {
                if let Some(ms) = to_u64_ms(v) { tx_interval_ms = Some(ms); }
            }
        }
        _ => {
            return Err("net:lldp => first arg must be string/array (iface) or keyed options".to_string());
        }
    }

    // Ensure iface is included in ifaces list for consistency
    if let Some(ref name) = iface {
        if !ifaces.iter().any(|x| x == name) {
            ifaces.push(name.clone());
        }
    }

    // Convert protocols
    let mut protocols: Vec<DiscoveryProtocol> = if proto_strs.is_empty() {
        vec![DiscoveryProtocol::LLDP]
    } else {
        proto_strs
            .iter()
            .filter_map(|s| DiscoveryProtocol::parse(s))
            .collect()
    };
    if protocols.is_empty() {
        protocols.push(DiscoveryProtocol::LLDP);
    }

    let effective_verbose = verbose_flag.unwrap_or_else(|| interp.is_verbose());

    // Build options and spawn iterator — **engine by default**, stub only if stub:true.
    let opts = LldpOptions {
        iface,
        ifaces,
        mode,
        protocols,
        ttl,
        count: count_opt,
        hostname,
        port_id,
        stub, // if true => use stub; default false => real engine
        verbose: effective_verbose,

        snaplen,
        promisc,
        capture_timeout_ms,
        channel_capacity,
        tx_interval_ms,
    }
    .normalized();

    let handle = spawn_iterator(opts);
    let tf = iter_to_transform(handle);
    Ok(Value::Function(tf))
}

/* ───────────────────────────── Cargo_lock registration ───────────────────────────── */

#[no_mangle]
pub unsafe extern "C" fn Cargo_lock(interp_ptr: *mut c_void, _extra_str: *const c_void) -> i32 {
    if interp_ptr.is_null() {
        return 1;
    }
    let interp = &mut *(interp_ptr as *mut Interpreter);

    // net:fetch
    let fetch_fn = Arc::new(Mutex::new(fetch_bridge_fn));
    interp.register_dynamic_function("net:fetch", fetch_fn);
    interp.set_variable(
        "net:fetch",
        Value::Function(Box::new(FunctionValue::Named("net:fetch".to_string()))),
    );

    // net:ping (one-shot transform that emits a KeyedArray per call)
    let ping_fn = Arc::new(Mutex::new(ping_bridge_fn));
    interp.register_dynamic_function("net:ping", ping_fn);
    interp.set_variable(
        "net:ping",
        Value::Function(Box::new(FunctionValue::Named("net:ping".to_string()))),
    );

    // net:real_ping (transform-producing raw ICMP stream; **API parity** with net:ping)
    let real_ping_fn = Arc::new(Mutex::new(real_ping_bridge_fn));
    interp.register_dynamic_function("net:real_ping", real_ping_fn);
    interp.set_variable(
        "net:real_ping",
        Value::Function(Box::new(FunctionValue::Named("net:real_ping".to_string()))),
    );

    // net:lldp (capture/parse)
    let lldp_fn = Arc::new(Mutex::new(lldp_bridge_fn));
    interp.register_dynamic_function("net:lldp", lldp_fn);
    interp.set_variable(
        "net:lldp",
        Value::Function(Box::new(FunctionValue::Named("net:lldp".to_string()))),
    );

    // Install the manager poller for fetch tasks.
    {
        let poller = Arc::new(Mutex::new(move |interp: &mut Interpreter| {
            let mut mgr = NET_MANAGER.lock().unwrap();
            mgr.poll_events(interp);
            ACTIVE_TASKS.load(std::sync::atomic::Ordering::SeqCst)
        }));
        interp.add_poller(poller);
    }

    0
}