net-mumu 0.2.0-rc.3

// src/lldp/engine.rs
// src/lldp/engine.rs
//
// Real LLDP/CDP engine (Linux listen mode) + event stream glue.
// ------------------------------------------------------------
// • Opens non-blocking AF_PACKET captures (one per interface), optionally
//   without promiscuous mode and with LLDP/CDP multicast joins.
// • Parses LLDP/CDP frames -> LldpRow (see parse.rs)
// • Maintains a NeighborTable for change detection and TTL expiry
// • Emits coalesced "add" / "update" and TTL-driven "remove" events
// • Returns:
//      Ok(Value)          — next event row (one per call)
//      Err("AGAIN")       — nothing to emit right now (keep polling)
//      Err("NO_MORE_DATA")— only when an explicit `count` cap was reached
//
// Non-Linux platforms: we return a single explicit error row (ok:false)
// explaining that the real engine is unavailable.
//

use indexmap::IndexMap;
use std::collections::VecDeque;

use mumu::parser::types::Value;

use super::iterator::{handle_from_engine, LldpEngine};
use super::options::{LldpMode, LldpOptions};
use super::proto::DiscoveryProtocol;
use super::table::{Event, NeighborTable};

#[inline]
fn now_ms() -> u64 {
    use std::time::{SystemTime, UNIX_EPOCH};
    SystemTime::now()
        .duration_since(UNIX_EPOCH)
        .unwrap_or_default()
        .as_millis() as u64
}

/// Convert a change Event + row payload into a KeyedArray Value that is
/// JSON-friendly and stable for Flow consumers.
///
/// Adds:
///   - ok: true
///   - event: "add" | "update" | "remove"
fn event_to_value(ev: Event) -> Value {
    let mut base = match ev.row.to_value() {
        Value::KeyedArray(m) => m,
        other => {
            let mut m = IndexMap::new();
            m.insert("ok".into(), Value::Bool(false));
            m.insert(
                "message".into(),
                Value::SingleString(format!("unexpected row type: {:?}", other)),
            );
            return Value::KeyedArray(m);
        }
    };

    // Prepend event metadata (IndexMap preserves insertion order)
    let mut out = IndexMap::new();
    out.insert("ok".into(), Value::Bool(true));
    out.insert(
        "event".into(),
        Value::SingleString(ev.kind.as_str().to_string()),
    );
    for (k, v) in base.drain(..) {
        out.insert(k, v);
    }
    Value::KeyedArray(out)
}

/* ───────────────────────────────────────────────────────────────────────────
   LINUX: Real engine implementation (listen mode)
   ───────────────────────────────────────────────────────────────────────── */

#[cfg(target_os = "linux")]
mod real {
    use super::*;
    use super::super::capture::{CaptureError, RawCapture};
    use super::super::parse::parse_any_row_from_frame;

    /// One capture context per interface.
    struct CaptureCtx {
        iface: String,
        cap: RawCapture,
        buf: Vec<u8>,
    }

    /// Live engine state.
    pub struct RealEngine {
        caps: Vec<CaptureCtx>,
        table: NeighborTable,
        queue: VecDeque<Value>,
        /// Last GC wall-clock in ms
        last_gc_ms: u64,
        /// GC cadence in ms (env: MUMU_LLDP_GC_MS, default 1000)
        gc_interval_ms: u64,
        /// Update coalesce window in ms (env: MUMU_LLDP_COALESCE_MS, default 1500)
        coalesce_ms: u64,
        /// Remaining events (if Some) before we terminate with NO_MORE_DATA
        remain: Option<usize>,
        /// Round-robin index
        round: usize,
        /// Verbose logging to stderr
        verbose: bool,
    }

    impl RealEngine {
        fn new(opts: LldpOptions, caps: Vec<CaptureCtx>) -> Self {
            let default_ttl_ms = (opts.ttl as u64).saturating_mul(1000).max(1000);
            let table = NeighborTable::new(default_ttl_ms);

            let gc_interval_ms = std::env::var("MUMU_LLDP_GC_MS")
                .ok()
                .and_then(|s| s.parse::<u64>().ok())
                .unwrap_or(1000);

            let coalesce_ms = std::env::var("MUMU_LLDP_COALESCE_MS")
                .ok()
                .and_then(|s| s.parse::<u64>().ok())
                .unwrap_or(1500);

            Self {
                caps,
                table,
                queue: VecDeque::new(),
                last_gc_ms: 0,
                gc_interval_ms,
                coalesce_ms,
                remain: opts.count,
                round: 0,
                verbose: opts.verbose,
            }
        }

        /// Decrement remain counter after emitting an event; return true if we must stop now.
        #[inline]
        fn note_emitted_and_is_done(&mut self) -> bool {
            if let Some(n) = self.remain {
                if n > 0 {
                    self.remain = Some(n - 1);
                }
            }
            matches!(self.remain, Some(0))
        }
    }

    impl LldpEngine for RealEngine {
        fn next(&mut self) -> Result<Value, String> {
            // Emit anything already queued first
            if let Some(v) = self.queue.pop_front() {
                let _done = self.note_emitted_and_is_done();
                return Ok(v);
            }

            // Round-robin over capture handles; attempt to read one frame
            let caps_len = self.caps.len();
            if caps_len > 0 {
                let start = self.round % caps_len;
                for i in 0..caps_len {
                    let idx = (start + i) % caps_len;
                    let ctx = &mut self.caps[idx];

                    match ctx.cap.recv(&mut ctx.buf) {
                        Ok(n) if n > 0 => {
                            let now = now_ms();
                            if let Some(row) =
                                parse_any_row_from_frame(&ctx.iface, &ctx.buf[..n], now)
                            {
                                if let Some(ev) = self
                                    .table
                                    .upsert_and_maybe_event(row, now, self.coalesce_ms)
                                {
                                    let v = event_to_value(ev);
                                    let _done = {
                                        let is_done_after = self.note_emitted_and_is_done();
                                        self.round = idx.wrapping_add(1);
                                        is_done_after
                                    };
                                    return Ok(v);
                                }
                            }
                            // If the frame was not LLDP/CDP, just continue.
                            self.round = idx.wrapping_add(1);
                            // Try next ctx in the same call to improve throughput modestly
                            continue;
                        }
                        Err(CaptureError::NoData) => {
                            // No data on this iface for now; try next
                            self.round = idx.wrapping_add(1);
                            continue;
                        }
                        Err(e) => {
                            if self.verbose {
                                eprintln!("[net:lldp] capture error on {} => {:?}", ctx.iface, e);
                            }
                            // Soft-ignore; next polls may still work.
                            self.round = idx.wrapping_add(1);
                            continue;
                        }
                        _ => {
                            self.round = idx.wrapping_add(1);
                            continue;
                        }
                    }
                }
            }

            // TTL expiry -> Remove events (periodic)
            let now = now_ms();
            if now.saturating_sub(self.last_gc_ms) >= self.gc_interval_ms {
                let removes = self.table.gc_expired_events(now);
                if !removes.is_empty() {
                    for ev in removes {
                        self.queue.push_back(event_to_value(ev));
                    }
                    self.last_gc_ms = now;
                    if let Some(v) = self.queue.pop_front() {
                        let _done = self.note_emitted_and_is_done();
                        return Ok(v);
                    }
                } else {
                    self.last_gc_ms = now;
                }
            }

            // If a hard event cap has been hit, terminate
            if matches!(self.remain, Some(0)) {
                return Err("NO_MORE_DATA".into());
            }

            // Nothing ready this tick
            Err("AGAIN".into())
        }
    }

    /// Public entry: spawn the **real** LLDP iterator (Linux only).
    pub fn spawn_iterator(opts: LldpOptions) -> mumu::parser::types::IteratorHandle {
        // Collect captures
        let mut caps: Vec<CaptureCtx> = Vec::new();
        let ifaces = {
            let v = opts.effective_ifaces();
            if v.is_empty() {
                vec!["eth0".to_string()]
            } else {
                v
            }
        };

        // Decide which protocol multicasts we care about.
        let want_lldp = opts
            .protocols
            .iter()
            .any(|p| *p == DiscoveryProtocol::LLDP);
        let want_cdp = opts
            .protocols
            .iter()
            .any(|p| *p == DiscoveryProtocol::CDP);

        for name in ifaces {
            // Use the configured capture opener: honor promiscuous choice and join multicast
            // groups when not in promiscuous mode.
            let opened = RawCapture::open_configured(Some(&name), opts.promisc, want_lldp, want_cdp);

            match opened {
                Ok(rc) => {
                    // Buffer sized from snaplen (with a small headroom)
                    let buf_cap = (opts.snaplen.max(64) as usize).saturating_add(64);
                    let cap = CaptureCtx {
                        iface: name,
                        cap: rc,
                        buf: vec![0u8; buf_cap],
                    };
                    if opts.verbose {
                        eprintln!(
                            "[net:lldp] opened capture iface='{}' promisc={} snaplen={} buf={}",
                            cap.iface, opts.promisc, opts.snaplen, buf_cap
                        );
                    }
                    caps.push(cap);
                }
                Err(e) => {
                    if opts.verbose {
                        eprintln!("[net:lldp] capture open failed on {} => {:?}", name, e);
                    }
                }
            }
        }

        if caps.is_empty() {
            // No captures: yield a single explanatory error row then EOF.
            let row = build_error_row_basic(
                &opts,
                "net:lldp — could not open any capture sockets (check permissions and interface names)",
            );
            return handle_from_engine(Box::new(ErrorOnceEngine { row: Some(row) }));
        }

        let eng = RealEngine::new(opts, caps);
        handle_from_engine(Box::new(eng))
    }

    /// Build a minimal error row (ok:false, message + context).
    fn build_error_row_basic(opts: &LldpOptions, message: &str) -> Value {
        let mut map: IndexMap<String, Value> = IndexMap::new();
        map.insert("ok".into(), Value::Bool(false));
        map.insert(
            "message".into(),
            Value::SingleString(message.to_string()),
        );
        let iface = if let Some(i) = &opts.iface {
            i.clone()
        } else if let Some(first) = opts.ifaces.get(0) {
            first.clone()
        } else {
            "unknown".to_string()
        };
        map.insert("iface".into(), Value::SingleString(iface));
        map.insert("protocol".into(), Value::SingleString("LLDP".into())); // default hint
        map.insert(
            "mode".into(),
            Value::SingleString(match opts.mode {
                LldpMode::Listen => "listen".into(),
                LldpMode::Advertise => "advertise".into(),
                LldpMode::Discover => "discover".into(),
            }),
        );
        Value::KeyedArray(map)
    }

    /// Engine that yields **one** error row (ok:false), then ends.
    struct ErrorOnceEngine {
        row: Option<Value>,
    }

    impl LldpEngine for ErrorOnceEngine {
        fn next(&mut self) -> Result<Value, String> {
            match self.row.take() {
                Some(v) => Ok(v),
                None => Err("NO_MORE_DATA".into()),
            }
        }
    }
}

/* ───────────────────────────────────────────────────────────────────────────
   NON-LINUX: Single explicit error row
   ───────────────────────────────────────────────────────────────────────── */

#[cfg(not(target_os = "linux"))]
mod real {
    use super::*;
    use crate::lldp::iterator::LldpEngine;

    pub fn spawn_iterator(opts: LldpOptions) -> mumu::parser::types::IteratorHandle {
        let row = build_error_row(&opts);
        handle_from_engine(Box::new(ErrorOnceEngine { row: Some(row) }))
    }

    fn build_error_row(opts: &LldpOptions) -> Value {
        let mut map: IndexMap<String, Value> = IndexMap::new();
        map.insert("ok".into(), Value::Bool(false));
        map.insert(
            "message".into(),
            Value::SingleString(
                "net:lldp real engine is unavailable on this platform (Linux-only)".to_string(),
            ),
        );
        let iface = if let Some(i) = &opts.iface {
            i.clone()
        } else if let Some(first) = opts.ifaces.get(0) {
            first.clone()
        } else {
            "unknown".to_string()
        };
        map.insert("iface".into(), Value::SingleString(iface));
        map.insert("protocol".into(), Value::SingleString("LLDP".into()));
        map.insert(
            "mode".into(),
            Value::SingleString(match opts.mode {
                LldpMode::Listen => "listen".into(),
                LldpMode::Advertise => "advertise".into(),
                LldpMode::Discover => "discover".into(),
            }),
        );
        Value::KeyedArray(map)
    }

    struct ErrorOnceEngine {
        row: Option<Value>,
    }

    impl LldpEngine for ErrorOnceEngine {
        fn next(&mut self) -> Result<Value, String> {
            match self.row.take() {
                Some(v) => Ok(v),
                None => Err("NO_MORE_DATA".into()),
            }
        }
    }
}

/* ───────────────────────────────────────────────────────────────────────────
   Public trampoline
   ───────────────────────────────────────────────────────────────────────── */

/// Public entry: spawn the real engine (platform-specific).
pub fn spawn_iterator(opts: LldpOptions) -> mumu::parser::types::IteratorHandle {
    real::spawn_iterator(opts)
}