net-mumu 0.2.0-rc.3

// src/lldp/table.rs
#![allow(dead_code)]

//
// Neighbor table for LLDP/CDP rows.
// ---------------------------------
// This module maintains a deduplicated, time-aware table of neighbors
// keyed by (iface, protocol, chassis_id, port_id). It provides:
//   • upsert()  — insert or update a row and classify the change
//   • gc_expired() — remove entries whose TTL has elapsed
//   • snapshot() / len() / clear()
//   • upsert_and_maybe_event() — change-coalesced **event** producer (add/update)
//   • gc_expired_events()      — removal **event** producer
//
// Event model
// -----------
// We produce *edge* events that downstream flows can consume:
//
//   EventKind::Add    — first time we see a neighbor key
//   EventKind::Update — attributes changed (system/port desc, mgmt IP, VLAN,
//                       caps, TTL, etc.), *coalesced* by a caller-provided window
//   EventKind::Remove — neighbor expired (TTL holdtime elapsed)
//
// Coalescing
// ----------
// To reduce flapping, callers pass `coalesce_ms` to `upsert_and_maybe_event`.
//  • Add: always emitted immediately.
//  • Update: emitted only if at least `coalesce_ms` has elapsed since the
//    last *emitted* event for that neighbor. Otherwise suppressed.
//  • Remove: produced by `gc_expired_events()` with no coalescing.
//
// TTLs
// ----
// LLDP/CDP report TTL in **seconds**. The table works in **milliseconds**.
// Callers pass `now_ms`; we compute `expires_at_ms` from row.ttl (or
// `default_ttl_ms` if TTL is absent).
//
// The table is single-threaded; wrap in a Mutex if you need concurrency.
//

use std::collections::HashMap;
use std::hash::{Hash, Hasher};

use super::proto::DiscoveryProtocol;
use super::row::LldpRow;

/// Global Hash impl for the protocol enum (local type ⇒ allowed).
impl Hash for DiscoveryProtocol {
    fn hash<H: Hasher>(&self, state: &mut H) {
        (*self as u8).hash(state);
    }
}

/// Unique key for a neighbor entry.
#[derive(Debug, Clone, Eq, PartialEq, Hash)]
pub struct NeighborKey {
    pub iface: String,
    pub protocol: DiscoveryProtocol,
    pub chassis_id: String,
    pub port_id: String,
}

impl NeighborKey {
    #[inline]
    pub fn from_row(row: &LldpRow) -> Self {
        Self {
            iface: row.interface.clone(),
            protocol: row.protocol,
            chassis_id: row.chassis_id.clone(),
            port_id: row.port_id.clone(),
        }
    }
}

/// Internal value stored per neighbor.
#[derive(Debug, Clone)]
struct NeighborEntry {
    row: LldpRow,
    /// Expiration deadline in milliseconds (unix-like origin not required).
    /// None means "never expire" (should be avoided; we always compute one).
    expires_at_ms: Option<u64>,
    /// Last time we incorporated this row (ms).
    last_seen_ms: u64,
    /// Last time we **emitted** an event for this neighbor (ms).
    /// Used to coalesce update noise.
    last_emitted_ms: u64,
}

/// Result of calling `upsert()`.
#[derive(Debug, Clone)]
pub enum UpsertDelta {
    /// First time we see this neighbor.
    Added(LldpRow),
    /// We already knew this neighbor, but its attributes changed.
    /// `before` is the previous row; `after` is the new one we stored.
    Updated { before: LldpRow, after: LldpRow },
    /// Nothing changed besides book-keeping (timestamp/TTL refresh).
    Unchanged,
}

/// Result for each removal during `gc_expired()`.
#[derive(Debug, Clone)]
pub struct Expired {
    pub key: NeighborKey,
    pub last_row: LldpRow,
}

/// Stream event kinds for downstream consumers.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum EventKind {
    Add,
    Update,
    Remove,
}

impl EventKind {
    #[inline]
    pub fn as_str(&self) -> &'static str {
        match self {
            EventKind::Add    => "add",
            EventKind::Update => "update",
            EventKind::Remove => "remove",
        }
    }
}

/// A change event with the full current row payload.
#[derive(Debug, Clone)]
pub struct Event {
    pub kind: EventKind,
    pub row: LldpRow,
}

/// A time-aware table of LLDP/CDP neighbors.
#[derive(Debug)]
pub struct NeighborTable {
    map: HashMap<NeighborKey, NeighborEntry>,
    /// Default TTL (milliseconds) used when a row carries no TTL.
    default_ttl_ms: u64,
}

impl NeighborTable {
    /// Create a new table with `default_ttl_ms` used when rows have no TTL.
    pub fn new(default_ttl_ms: u64) -> Self {
        Self {
            map: HashMap::new(),
            default_ttl_ms: default_ttl_ms.max(1),
        }
    }

    /// Number of currently tracked neighbors.
    #[inline]
    pub fn len(&self) -> usize {
        self.map.len()
    }

    /// True if there are no neighbors tracked.
    #[inline]
    pub fn is_empty(&self) -> bool {
        self.map.is_empty()
    }

    /// Remove everything.
    pub fn clear(&mut self) {
        self.map.clear();
    }

    /// Return a vector of the current rows (unordered snapshot).
    pub fn snapshot(&self) -> Vec<LldpRow> {
        self.map.values().map(|e| e.row.clone()).collect()
    }

    /// Compute the (re)calculated expiry moment for a row.
    fn compute_expires_at_ms(&self, row: &LldpRow, now_ms: u64) -> u64 {
        let ttl_s: u64 = row.ttl.map(|t| t as u64).unwrap_or_else(|| self.default_ttl_ms / 1000);
        let ttl_ms = if row.ttl.is_some() {
            ttl_s.saturating_mul(1000)
        } else {
            self.default_ttl_ms
        };
        now_ms.saturating_add(ttl_ms.max(1))
    }

    /// Insert or update an entry and return the delta classification.
    ///
    /// This refreshes the expiration based on the row's TTL (or default TTL).
    /// `timestamp_ms` in the row is not used for change detection.
    pub fn upsert(&mut self, row: LldpRow, now_ms: u64) -> UpsertDelta {
        let key = NeighborKey::from_row(&row);
        let expires_at = self.compute_expires_at_ms(&row, now_ms);

        match self.map.get_mut(&key) {
            None => {
                let entry = NeighborEntry {
                    row: row.clone(),
                    expires_at_ms: Some(expires_at),
                    last_seen_ms: now_ms,
                    last_emitted_ms: 0,
                };
                self.map.insert(key, entry);
                UpsertDelta::Added(row)
            }
            Some(entry) => {
                let before = entry.row.clone();
                entry.last_seen_ms = now_ms;
                entry.expires_at_ms = Some(expires_at);

                let changed = rows_meaningfully_differ(&before, &row);

                entry.row = row.clone();
                if changed {
                    UpsertDelta::Updated { before, after: row }
                } else {
                    UpsertDelta::Unchanged
                }
            }
        }
    }

    /// Garbage-collect entries whose TTL has elapsed at `now_ms`.
    /// Returns the list of expired entries (with their last known rows).
    pub fn gc_expired(&mut self, now_ms: u64) -> Vec<Expired> {
        let mut expired: Vec<Expired> = Vec::new();

        // Collect keys to remove before mutating the map while iterating.
        let keys_to_remove: Vec<NeighborKey> = self
            .map
            .iter()
            .filter_map(|(k, v)| match v.expires_at_ms {
                Some(deadline) if deadline <= now_ms => Some(k.clone()),
                _ => None,
            })
            .collect();

        for k in keys_to_remove {
            if let Some(entry) = self.map.remove(&k) {
                expired.push(Expired {
                    key: k,
                    last_row: entry.row,
                });
            }
        }

        expired
    }

    /// Get the current row for a neighbor key.
    #[inline]
    pub fn get(&self, key: &NeighborKey) -> Option<&LldpRow> {
        self.map.get(key).map(|e| &e.row)
    }

    /// Remove a specific neighbor and return its last row (if present).
    pub fn remove(&mut self, key: &NeighborKey) -> Option<LldpRow> {
        self.map.remove(key).map(|e| e.row)
    }

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

    /// Upsert a row and return a *coalesced* change event, if any.
    ///
    /// Arguments:
    ///   • row          — new sample
    ///   • now_ms       — current time in milliseconds
    ///   • coalesce_ms  — minimum elapsed time since *last emitted event* for
    ///                     this neighbor before we emit an Update event
    ///
    /// Returns:
    ///   Some(Event { Add, row })     — on first sighting
    ///   Some(Event { Update, row })  — on meaningful change and coalesce window elapsed
    ///   None                         — on Unchanged *or* Update suppressed by coalesce
    pub fn upsert_and_maybe_event(
        &mut self,
        row: LldpRow,
        now_ms: u64,
        coalesce_ms: u64,
    ) -> Option<Event> {
        let key = NeighborKey::from_row(&row);
        match self.upsert(row, now_ms) {
            UpsertDelta::Added(new_row) => {
                if let Some(e) = self.map.get_mut(&key) {
                    e.last_emitted_ms = now_ms;
                }
                Some(Event { kind: EventKind::Add, row: new_row })
            }
            UpsertDelta::Updated { after: new_row, .. } => {
                let entry = self.map.get_mut(&key)?;
                // Coalesce: only emit if sufficient time elapsed since last *emitted* event
                if entry.last_emitted_ms == 0 || now_ms.saturating_sub(entry.last_emitted_ms) >= coalesce_ms {
                    entry.last_emitted_ms = now_ms;
                    Some(Event { kind: EventKind::Update, row: new_row })
                } else {
                    None
                }
            }
            UpsertDelta::Unchanged => None,
        }
    }

    /// Run TTL GC and return **Remove** events for all expired neighbors.
    #[inline]
    pub fn gc_expired_events(&mut self, now_ms: u64) -> Vec<Event> {
        self.gc_expired(now_ms)
            .into_iter()
            .map(|ex| Event { kind: EventKind::Remove, row: ex.last_row })
            .collect()
    }
}

/* ───────────────────────────── Change detection ───────────────────────────── */

/// Two rows "meaningfully differ" if any field other than `timestamp_ms`
/// (and ignoring superficial whitespace differences) has changed.
///
/// Notes:
///   • TTL changes are *considered* a real update.
///   • Capability sets are compared as unordered (by normalizing/sorting).
fn rows_meaningfully_differ(a: &LldpRow, b: &LldpRow) -> bool {
    if a.interface != b.interface {
        return true;
    }
    if a.protocol != b.protocol {
        return true;
    }
    if norm(&a.chassis_id) != norm(&b.chassis_id) {
        return true;
    }
    if norm(&a.port_id) != norm(&b.port_id) {
        return true;
    }

    if strip_opt(&a.system_name) != strip_opt(&b.system_name) {
        return true;
    }
    if strip_opt(&a.system_desc) != strip_opt(&b.system_desc) {
        return true;
    }
    if strip_opt(&a.port_desc) != strip_opt(&b.port_desc) {
        return true;
    }

    if strip_opt(&a.management_ip) != strip_opt(&b.management_ip) {
        return true;
    }
    if strip_opt(&a.vlan) != strip_opt(&b.vlan) {
        return true;
    }

    if a.ttl != b.ttl {
        return true;
    }

    let mut ca = a.capabilities.iter().map(|s| norm(s)).collect::<Vec<_>>();
    let mut cb = b.capabilities.iter().map(|s| norm(s)).collect::<Vec<_>>();
    ca.sort();
    cb.sort();
    if ca != cb {
        return true;
    }

    false
}

#[inline]
fn norm(s: &str) -> String {
    s.trim().to_ascii_lowercase()
}

#[inline]
fn strip_opt(s: &Option<String>) -> Option<String> {
    s.as_ref().map(|x| norm(x))
}

/* ───────────────────────────── Tests (basic) ───────────────────────────── */

#[cfg(test)]
mod tests {
    use super::*;
    use crate::lldp::proto::DiscoveryProtocol;

    fn base_row(iface: &str) -> LldpRow {
        LldpRow {
            interface: iface.into(),
            protocol: DiscoveryProtocol::LLDP,
            chassis_id: "00:11:22:33:44:55".into(),
            port_id: "Gi1/0/1".into(),
            system_name: Some("sw-1".into()),
            system_desc: Some("stub".into()),
            port_desc: Some("uplink".into()),
            vlan: Some("10".into()),
            management_ip: Some("192.0.2.10".into()),
            capabilities: vec!["bridge".into(), "router".into()],
            ttl: Some(120),
            timestamp_ms: 12345,
        }
    }

    #[test]
    fn add_then_unchanged() {
        let mut t = NeighborTable::new(30_000);
        let now = 1_000_000;

        let r1 = base_row("eth0");
        match t.upsert(r1.clone(), now) {
            UpsertDelta::Added(_) => {}
            other => panic!("expected Added, got {:?}", other),
        }
        assert_eq!(t.len(), 1);

        let mut r2 = r1.clone();
        r2.timestamp_ms = now + 777;
        match t.upsert(r2, now + 1_000) {
            UpsertDelta::Unchanged => {}
            other => panic!("expected Unchanged, got {:?}", other),
        }
        assert_eq!(t.len(), 1);
    }

    #[test]
    fn add_then_update_on_cap_change() {
        let mut t = NeighborTable::new(30_000);
        let now = 1_000_000;

        let r1 = base_row("eth0");
        let _ = t.upsert(r1.clone(), now);

        let mut r2 = r1.clone();
        r2.capabilities = vec!["bridge".into()];
        match t.upsert(r2.clone(), now + 10_000) {
            UpsertDelta::Updated { before, after } => {
                assert_eq!(before.management_ip, Some("192.0.2.10".into()));
                assert_eq!(after.capabilities, vec!["bridge".to_string()]);
            }
            other => panic!("expected Updated, got {:?}", other),
        }
    }

    #[test]
    fn expiry_removes_entry() {
        let mut t = NeighborTable::new(10_000);
        let now = 1_000_000;

        let mut r = base_row("eth0");
        r.ttl = Some(5);
        let _ = t.upsert(r.clone(), now);

        let expired = t.gc_expired(now + 4_999);
        assert!(expired.is_empty());
        assert_eq!(t.len(), 1);

        let expired = t.gc_expired(now + 5_001);
        assert_eq!(expired.len(), 1);
        assert!(t.is_empty());
    }

    #[test]
    fn event_coalescing_update() {
        let mut t = NeighborTable::new(60_000);
        let mut now = 1_000_000;

        // Add => event emitted
        let r1 = base_row("eth0");
        let ev = t.upsert_and_maybe_event(r1.clone(), now, 1_500).expect("add event");
        assert_eq!(ev.kind, EventKind::Add);

        // Small change but within coalesce window ⇒ no event
        now += 500;
        let mut r2 = r1.clone();
        r2.port_desc = Some("uplink-A".into());
        assert!(t.upsert_and_maybe_event(r2, now, 1_500).is_none());

        // After coalesce window ⇒ update event
        now += 2_000;
        let mut r3 = r1.clone();
        r3.port_desc = Some("uplink-B".into());
        let ev2 = t.upsert_and_maybe_event(r3, now, 1_500).expect("update event");
        assert_eq!(ev2.kind, EventKind::Update);
    }

    #[test]
    fn gc_expired_events_remove() {
        let mut t = NeighborTable::new(10_000);
        let now = 1_000_000;

        let mut r = base_row("eth1");
        r.ttl = Some(5);
        let _ = t.upsert(r.clone(), now);

        // advance beyond TTL
        let evs = t.gc_expired_events(now + 6_000);
        assert_eq!(evs.len(), 1);
        assert_eq!(evs[0].kind, EventKind::Remove);
        assert!(t.is_empty());
    }
}