net-mumu 0.2.0-rc.3

// src/lldp/stub.rs
//
// Synthetic LLDP/CDP stream (opt-in).
// -----------------------------------
// This stub is **only** used when explicitly requested (opts.stub == true)
// or when the environment variable `MUMU_LLDP_USE_STUB` is set (see mod.rs).
//
// What it does
// ------------
// • Produces a Flow-friendly, 0-arg transform (via iterator glue) that yields
//   **one event row per call**.
// • Each row matches the *engine* schema and includes:
//       ok: true,
//       event: "add" | "update",
//       iface, protocol, chassis_id, port_id,
//       system_name?, system_desc?, port_desc?,
//       vlan?, management_ip?, capabilities[], ttl?, timestamp_ms
// • Multiple neighbors are synthesized by rotating through the provided
//   interfaces and protocols. The first sighting per neighbor is "add";
//   subsequent sightings are "update" (with a tiny change every few pulls).
//
// Why keep this around?
// ---------------------
// • For development & CI: lets you exercise Flow graphs without requiring
//   raw capture privileges or hardware. The *real* engine is still the default.
// • For demos / docs: predictable output and cadence.
//
// (c) 2025 MuMu contributors — MIT/Apache-2.0

use std::collections::HashSet;
use std::time::{SystemTime, UNIX_EPOCH};

use indexmap::IndexMap;
use mumu::parser::types::Value;

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

/// Simple composite key for "first sighting" detection.
#[derive(Hash, Eq, PartialEq)]
struct SeenKey {
    iface: String,
    proto: DiscoveryProtocol,
    chassis_id: String,
    port_id: String,
}

impl SeenKey {
    fn new(iface: &str, proto: DiscoveryProtocol, chassis_id: &str, port_id: &str) -> Self {
        Self {
            iface: iface.to_string(),
            proto,
            chassis_id: chassis_id.to_string(),
            port_id: port_id.to_string(),
        }
    }
}

/// A streaming engine: generates one synthetic **event** row per `next()` call.
/// Honors `count` if provided; otherwise unbounded.
struct StreamingEngine {
    _mode: LldpMode,
    ifaces: Vec<String>,
    protocols: Vec<DiscoveryProtocol>,
    ttl: u16,
    hostname: String,
    port_id_override: Option<String>,
    remain: Option<usize>,
    idx: usize,
    verbose: bool,
    seen: HashSet<SeenKey>,
    update_period: usize,
}

impl StreamingEngine {
    fn new(opts: LldpOptions) -> Self {
        // Interface selection (prefer explicit iface, else any provided, else default)
        let mut ifaces: Vec<String> = if !opts.ifaces.is_empty() {
            opts.ifaces.clone()
        } else if let Some(primary) = opts.iface.clone() {
            vec![primary]
        } else {
            vec!["eth0".to_string()]
        };
        // Ensure `iface` (if set) is included
        if let Some(primary) = &opts.iface {
            if !ifaces.iter().any(|x| x == primary) {
                ifaces.push(primary.clone());
            }
        }

        // Protocols default to LLDP if none provided
        let protocols: Vec<DiscoveryProtocol> = if opts.protocols.is_empty() {
            vec![DiscoveryProtocol::LLDP]
        } else {
            opts.protocols.clone()
        };

        Self {
            _mode: opts.mode,
            ifaces,
            protocols,
            ttl: opts.ttl.max(1),
            hostname: opts.hostname.unwrap_or_else(|| "stub-host".into()),
            port_id_override: opts.port_id,
            remain: opts.count, // None => unbounded
            idx: 0,
            verbose: opts.verbose,
            seen: HashSet::new(),
            update_period: 8, // generate a small attribute change every 8th emission
        }
    }

    /// Build a deterministic pseudo-neighbor for a given counter `i`.
    fn synth_values(&self, i: usize) -> (String, DiscoveryProtocol, String, String, String, String, String, Vec<String>) {
        let iface = self.ifaces[i % self.ifaces.len()].clone();
        let proto = self.protocols[(i / self.ifaces.len()) % self.protocols.len()];

        // Deterministic pseudo-values so successive rows vary but remain stable-ish for a given i.
        let oct = 0x10u8.wrapping_add((i as u8) % 0xEF);
        let chassis_id = format!("aa:bb:cc:dd:ee:{:02x}", oct);

        let port_id = self
            .port_id_override
            .clone()
            .unwrap_or_else(|| iface.clone());

        let system_name = self.hostname.clone();
        let system_desc = "MuMu-LLDP stub 0.1".to_string();

        // Vary management IP in a limited range to look realistic.
        let management_ip = format!("198.51.100.{}", 10 + (i % 140));

        // Start with two common caps.
        let capabilities = vec!["bridge".into(), "router".into()];

        (iface, proto, chassis_id, port_id, system_name, system_desc, management_ip, capabilities)
    }

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

impl LldpEngine for StreamingEngine {
    fn next(&mut self) -> Result<Value, String> {
        // Respect count limit if present
        if let Some(left) = self.remain.as_mut() {
            if *left == 0 {
                return Err("NO_MORE_DATA".into());
            }
        }

        let i = self.idx;
        self.idx = self.idx.wrapping_add(1);

        let (iface, proto, chassis_id, mut port_id, system_name, mut system_desc, management_ip, capabilities) =
            self.synth_values(i);

        // Change port_desc and/or VLAN periodically to simulate updates.
        let mut port_desc = format!("StubPort {}", iface);
        let vlan = if (i % self.update_period) == 0 && i != 0 {
            // mutate fields to create a meaningful update
            port_desc = format!("StubPort {} (u{})", iface, i);
            system_desc = format!("MuMu-LLDP stub 0.1 | u{}", i);
            // Alternate port_id slightly if no override was given, to show variety
            if self.port_id_override.is_none() {
                port_id = format!("{}:{}", port_id, (i % 4) + 1);
            }
            Some(((i % 4094) + 1).to_string())
        } else {
            None
        };

        // Determine event kind (add first time per neighbor key, otherwise update)
        let key = SeenKey::new(&iface, proto, &chassis_id, &port_id);
        let is_new = self.seen.insert(key);

        // Prepare event row (IndexMap preserves insertion order for "ok", "event", then payload)
        let mut out: IndexMap<String, Value> = IndexMap::new();
        out.insert("ok".into(), Value::Bool(true));
        out.insert(
            "event".into(),
            Value::SingleString(if is_new { "add".into() } else { "update".into() }),
        );

        // Base payload — match **engine** schema (no legacy keys)
        out.insert("iface".into(), Value::SingleString(iface));
        out.insert(
            "protocol".into(),
            Value::SingleString(proto.as_str().to_string()),
        );
        out.insert("chassis_id".into(), Value::SingleString(chassis_id));
        out.insert("port_id".into(), Value::SingleString(port_id));

        out.insert("system_name".into(), Value::SingleString(system_name));
        out.insert("system_desc".into(), Value::SingleString(system_desc));
        out.insert("port_desc".into(), Value::SingleString(port_desc));
        if let Some(v) = vlan {
            out.insert("vlan".into(), Value::SingleString(v));
        }
        out.insert(
            "management_ip".into(),
            Value::SingleString(management_ip),
        );
        out.insert("capabilities".into(), Value::StrArray(capabilities));
        out.insert("ttl".into(), Value::Int(self.ttl as i32));
        out.insert("timestamp_ms".into(), Value::Long(Self::now_ms() as i64));

        if self.verbose {
            if let Value::SingleString(ev) = out.get("event").cloned().unwrap_or(Value::SingleString("?".into())) {
                if let Value::SingleString(proto_s) = out
                    .get("protocol")
                    .cloned()
                    .unwrap_or(Value::SingleString("?".into()))
                {
                    if let Value::SingleString(ifname) = out
                        .get("iface")
                        .cloned()
                        .unwrap_or(Value::SingleString("?".into()))
                    {
                        eprintln!(
                            "[net:lldp][stub] emit event={} proto={} iface={}",
                            ev, proto_s, ifname
                        );
                    }
                }
            }
        }

        // Decrement remaining if bounded
        if let Some(left) = self.remain.as_mut() {
            if *left > 0 {
                *left -= 1;
            }
        }

        Ok(Value::KeyedArray(out))
    }
}

/// Create an iterator handle backed by the **streaming** stub engine.
/// `bridge.rs` calls this with a single `LldpOptions` argument.
pub fn spawn_iterator(opts: LldpOptions) -> mumu::parser::types::IteratorHandle {
    if opts.verbose {
        eprintln!(
            "[net:lldp] stub spawn: mode={:?} iface={:?} ifaces={:?} ttl={} count={:?} protos={:?} hostname={:?} port_id={:?}",
            opts.mode,
            opts.iface,
            if opts.ifaces.is_empty() { None::<Vec<String>> } else { Some(opts.ifaces.clone()) },
            opts.ttl,
            opts.count,
            if opts.protocols.is_empty() { vec![DiscoveryProtocol::LLDP] } else { opts.protocols.clone() },
            opts.hostname,
            opts.port_id,
        );
    }

    let engine = StreamingEngine::new(opts);
    handle_from_engine(Box::new(engine))
}