use chrono::{DateTime, Duration, Utc};
use std::collections::{HashMap, HashSet, VecDeque};

use super::types::*;

/// In-memory knowledge graph with O(1) indexes for all critical queries.
pub struct KnowledgeGraph {
    // ── Storage ──
    pub(crate) nodes: HashMap<NodeId, Node>,
    pub(crate) edges: Vec<Edge>,

    // ── Dedup indexes (entity → NodeId) ──
    pub(crate) pid_index: HashMap<u32, NodeId>,
    pub(crate) ip_index: HashMap<String, NodeId>,
    pub(crate) file_index: HashMap<String, NodeId>,
    pub(crate) user_index: HashMap<String, NodeId>,
    pub(crate) domain_index: HashMap<String, NodeId>,
    pub(crate) port_index: HashMap<(u16, String), NodeId>,
    pub(crate) container_index: HashMap<String, NodeId>,
    pub(crate) device_index: HashMap<String, NodeId>, // "vendor:product:serial"
    pub(crate) incident_index: HashMap<String, NodeId>,
    pub(crate) campaign_index: HashMap<String, NodeId>,
    pub(crate) system_node: Option<NodeId>,

    // ── Adjacency lists (NodeId → edge indexes) ──
    pub(crate) outgoing: HashMap<NodeId, Vec<usize>>,
    pub(crate) incoming: HashMap<NodeId, Vec<usize>>,

    /// Per-node "most recent edge timestamp", used by `enforce_memory_limit`
    /// to evict by LRU without paying O(E_avg log E_avg) per node. Updated on
    /// every `add_edge`, cleared on `remove_node`. Rebuilt from `edges` on
    /// `reconstruct_from_snapshot` (same precedent as `outgoing`/`incoming`),
    /// so it does not enter the persistence wire format.
    pub(crate) last_edge_ts: HashMap<NodeId, DateTime<Utc>>,

    // ── Fast-access sets ──
    pub(crate) threat_intel_nodes: HashSet<NodeId>,

    // ── Counters ──
    pub(crate) next_id: NodeId,
    pub(crate) memory_estimate: usize,
    pub(crate) tombstoned_edges: usize,
    pub max_memory: usize,
    pub created_at: DateTime<Utc>,

    // ── Event telemetry (for sensors tab) ──
    //
    // Wave 6c (memory-opt, 2026-05-05): all three keyed collections
    // here are populated per event via `record_event_telemetry`. Keys
    // are low-cardinality (`source` ~10 distinct, `kind` ~50 distinct,
    // bucket date ~24 distinct/day) but the per-event `.entry(...)`
    // calls were pre-Wave-6c allocating fresh `String` PER EVENT. With
    // `Arc<str>` interned at the call site, every repeat is a 16-byte
    // pointer-clone. Wire format on snapshot is unchanged
    // (serde renders `Arc<str>` as plain JSON string).
    pub(crate) source_counts: HashMap<std::sync::Arc<str>, usize>,
    pub(crate) kind_counts: HashMap<std::sync::Arc<str>, usize>,
    pub(crate) event_timeline:
        std::collections::BTreeMap<std::sync::Arc<str>, HashMap<std::sync::Arc<str>, usize>>,
    pub(crate) total_events_ingested: usize,

    // ── Transient: current event metadata (set during ingest, used by add_edge) ──
    //
    // Wave 6 (memory-opt, 2026-05-05): `_current_event_source` and
    // `_current_event_kind` are `Arc<str>` interned at the ingest site
    // so the per-event clone is a 16-byte pointer bump instead of a
    // fresh String allocation. Per-event matters: a busy host produces
    // hundreds of ingest calls per second; pre-Wave-6 each call paid
    // 2 × String alloc on these two fields. `summary` and `severity`
    // stay `String` because they are unique-per-event (summary) or
    // already cheap (severity is a 4-7 byte word).
    pub(crate) _current_event_source: Option<std::sync::Arc<str>>,
    pub(crate) _current_event_kind: Option<std::sync::Arc<str>>,
    pub(crate) _current_event_summary: Option<String>,
    pub(crate) _current_event_severity: Option<String>,

    // ── Real-time trigger state ──
    /// Incidents generated by edge-insert triggers (collected during ingest, drained by caller).
    pub(crate) trigger_incidents: Vec<innerwarden_core::incident::Incident>,
    /// Host label for trigger incident generation.
    pub(crate) trigger_host: String,
}

impl KnowledgeGraph {
    pub fn new() -> Self {
        Self {
            nodes: HashMap::new(),
            edges: Vec::new(),
            pid_index: HashMap::new(),
            ip_index: HashMap::new(),
            file_index: HashMap::new(),
            user_index: HashMap::new(),
            domain_index: HashMap::new(),
            port_index: HashMap::new(),
            container_index: HashMap::new(),
            device_index: HashMap::new(),
            incident_index: HashMap::new(),
            campaign_index: HashMap::new(),
            system_node: None,
            outgoing: HashMap::new(),
            incoming: HashMap::new(),
            last_edge_ts: HashMap::new(),
            threat_intel_nodes: HashSet::new(),
            next_id: 1,
            memory_estimate: 0,
            tombstoned_edges: 0,
            max_memory: 50 * 1024 * 1024, // 50 MB
            source_counts: HashMap::new(),
            kind_counts: HashMap::new(),
            event_timeline: std::collections::BTreeMap::new(),
            total_events_ingested: 0,
            _current_event_source: None,
            _current_event_kind: None,
            _current_event_summary: None,
            _current_event_severity: None,
            created_at: Utc::now(),
            trigger_incidents: Vec::new(),
            trigger_host: String::new(),
        }
    }

    /// Set the host label used for trigger incidents.
    pub fn set_trigger_host(&mut self, host: &str) {
        self.trigger_host = host.to_string();
    }

    /// Drain accumulated trigger incidents (call after ingest batch).
    pub fn drain_trigger_incidents(&mut self) -> Vec<innerwarden_core::incident::Incident> {
        std::mem::take(&mut self.trigger_incidents)
    }

    // ── Node CRUD ───────────────────────────────────────────────────────

    /// Add a node, returning its ID. If a node with the same dedup key exists,
    /// returns the existing ID without creating a duplicate.
    pub fn add_node(&mut self, node: Node) -> NodeId {
        // Check dedup
        if let Some(existing) = self.find_existing(&node) {
            return existing;
        }

        let id = self.next_id;
        self.next_id += 1;

        self.index_node(id, &node);
        self.memory_estimate += Self::estimate_node_size(&node);
        self.nodes.insert(id, node);

        id
    }

    /// Add or update: if the dedup key exists, update the node in-place and return its ID.
    /// Otherwise create a new node.
    pub fn upsert_node(&mut self, node: Node) -> NodeId {
        if let Some(existing_id) = self.find_existing(&node) {
            // Update the existing node with new data
            if let Some(existing) = self.nodes.get_mut(&existing_id) {
                Self::merge_node(existing, &node);
            }
            // Update threat intel index if needed
            if let Node::Ip { ref datasets, .. } = node {
                if !datasets.is_empty() {
                    self.threat_intel_nodes.insert(existing_id);
                }
            }
            existing_id
        } else {
            self.add_node(node)
        }
    }

    pub fn get_node(&self, id: NodeId) -> Option<&Node> {
        self.nodes.get(&id)
    }

    pub fn get_node_mut(&mut self, id: NodeId) -> Option<&mut Node> {
        self.nodes.get_mut(&id)
    }

    pub fn remove_node(&mut self, id: NodeId) {
        if let Some(node) = self.nodes.remove(&id) {
            self.memory_estimate = self
                .memory_estimate
                .saturating_sub(Self::estimate_node_size(&node));
            self.deindex_node(id, &node);
            self.threat_intel_nodes.remove(&id);
            self.last_edge_ts.remove(&id);

            // Collect edge indexes to tombstone
            let mut edge_idxs: Vec<usize> = self.outgoing.remove(&id).unwrap_or_default();
            edge_idxs.extend(self.incoming.remove(&id).unwrap_or_default());
            edge_idxs.sort_unstable();
            edge_idxs.dedup();

            for idx in &edge_idxs {
                if *idx < self.edges.len() {
                    let edge = &self.edges[*idx];
                    let other = if edge.from == id { edge.to } else { edge.from };
                    if let Some(adj) = self.outgoing.get_mut(&other) {
                        adj.retain(|&i| i != *idx);
                    }
                    if let Some(adj) = self.incoming.get_mut(&other) {
                        adj.retain(|&i| i != *idx);
                    }
                }
                self.tombstoned_edges += 1;
            }
        }
    }

    /// Compact the edge vec by removing tombstoned entries and rebuilding adjacency.
    /// Called periodically when tombstone ratio exceeds 20%.
    ///
    /// 2026-05-03 (Wave 5b PR-4): the 20% ratio gate exists to amortise
    /// the O(E) rebuild cost across many remove_node calls. It is the
    /// right behaviour during steady-state ingest. It is NOT the right
    /// behaviour right before a snapshot serialise — tombstoned edges
    /// would leak into the persisted blob and reload would tag them
    /// as "dangling" (operator-visible WARN every time the snapshot
    /// is round-tripped). The slow_loop path now calls
    /// `compact_edges_force` instead; this gated method is retained
    /// as the canonical inline-mutation entry point for any future
    /// site that needs amortised compaction without forcing a
    /// full rebuild on every call.
    #[allow(dead_code)]
    pub fn compact_edges(&mut self) {
        if self.edges.is_empty() || self.tombstoned_edges == 0 {
            return;
        }
        let ratio = self.tombstoned_edges as f32 / self.edges.len() as f32;
        if ratio < 0.2 {
            return;
        }

        // Collect live edges (those still referenced by adjacency lists)
        let mut live: HashSet<usize> = HashSet::new();
        for idxs in self.outgoing.values() {
            for &i in idxs {
                live.insert(i);
            }
        }

        let mut new_edges = Vec::with_capacity(live.len());
        let mut old_to_new: HashMap<usize, usize> = HashMap::new();

        for (old_idx, edge) in self.edges.iter().enumerate() {
            if live.contains(&old_idx) {
                let new_idx = new_edges.len();
                old_to_new.insert(old_idx, new_idx);
                new_edges.push(edge.clone());
            } else {
                self.memory_estimate = self
                    .memory_estimate
                    .saturating_sub(Self::estimate_edge_size(edge));
            }
        }

        // Remap adjacency lists
        for idxs in self.outgoing.values_mut() {
            *idxs = idxs
                .iter()
                .filter_map(|i| old_to_new.get(i).copied())
                .collect();
        }
        for idxs in self.incoming.values_mut() {
            *idxs = idxs
                .iter()
                .filter_map(|i| old_to_new.get(i).copied())
                .collect();
        }

        let removed = self.edges.len() - new_edges.len();
        self.edges = new_edges;
        self.tombstoned_edges = 0;

        if removed > 0 {
            tracing::debug!(
                removed,
                live = self.edges.len(),
                "knowledge graph: compacted edges"
            );
        }
    }

    /// 2026-05-03 (Wave 5b PR-4): unconditional compaction for the
    /// pre-serialise path. Same body as `compact_edges` minus the
    /// 20%-ratio gate. Use this from slow_loop right before
    /// `serialize_snapshot_bytes` so tombstoned edges accumulated by
    /// `enforce_memory_limit` and `cleanup_expired` do not leak into
    /// the persisted blob.
    ///
    /// Why a separate method instead of dropping the gate from
    /// `compact_edges`: the gated path is the right behaviour during
    /// inline mutations (avoid O(E) work on every remove). The
    /// unconditional path is the right behaviour when the next
    /// operation is a serialise — we are about to walk every edge
    /// anyway, so paying compaction now amortises into the same scan.
    pub fn compact_edges_force(&mut self) {
        if self.edges.is_empty() || self.tombstoned_edges == 0 {
            return;
        }

        // Same body as `compact_edges` from here on. Kept as a
        // dedicated method (not a refactor that calls into a shared
        // helper) because the gate is the only difference and
        // splitting further would obscure the call graph the heap-
        // budget anchor walks.
        let mut live: HashSet<usize> = HashSet::new();
        for idxs in self.outgoing.values() {
            for &i in idxs {
                live.insert(i);
            }
        }

        let mut new_edges = Vec::with_capacity(live.len());
        let mut old_to_new: HashMap<usize, usize> = HashMap::new();

        for (old_idx, edge) in self.edges.iter().enumerate() {
            if live.contains(&old_idx) {
                let new_idx = new_edges.len();
                old_to_new.insert(old_idx, new_idx);
                new_edges.push(edge.clone());
            } else {
                self.memory_estimate = self
                    .memory_estimate
                    .saturating_sub(Self::estimate_edge_size(edge));
            }
        }

        for idxs in self.outgoing.values_mut() {
            *idxs = idxs
                .iter()
                .filter_map(|i| old_to_new.get(i).copied())
                .collect();
        }
        for idxs in self.incoming.values_mut() {
            *idxs = idxs
                .iter()
                .filter_map(|i| old_to_new.get(i).copied())
                .collect();
        }

        let removed = self.edges.len() - new_edges.len();
        self.edges = new_edges;
        self.tombstoned_edges = 0;

        if removed > 0 {
            tracing::debug!(
                removed,
                live = self.edges.len(),
                "knowledge graph: force-compacted edges before snapshot"
            );
        }
    }

    // ── Edge CRUD ───────────────────────────────────────────────────────

    /// Add an edge. Edges are never deduplicated — each represents a discrete event.
    /// If event metadata is set (during ingest), automatically adds source/kind/summary.
    pub fn add_edge(&mut self, mut edge: Edge) -> usize {
        // Enrich edge with current event metadata if available. All
        // four hot keys go through the interner (per-key) so the
        // 145k+ edges in prod share four `Arc<str>` allocations
        // instead of paying a fresh `String` per insert.
        use crate::knowledge_graph::intern::intern;
        if let Some(ref src) = self._current_event_source {
            if !edge.properties.contains_key("event_source") {
                // Wave 6: src is now `Arc<str>` — deref to `&str` via
                // `&**` for the serde_json::Value::from. The unstable
                // `Arc::<str>::as_str` (feature gate `str_as_str`) is
                // not available on stable, so this is the right form.
                edge.properties
                    .insert(intern("event_source"), serde_json::Value::from(&**src));
            }
        }
        if let Some(ref kind) = self._current_event_kind {
            if !edge.properties.contains_key("event_kind") {
                edge.properties
                    .insert(intern("event_kind"), serde_json::Value::from(&**kind));
            }
        }
        if let Some(ref summary) = self._current_event_summary {
            if !summary.is_empty() && !edge.properties.contains_key("summary") {
                // Truncate summary to save memory. Wave 1 (AUDIT-WAVE1-UTF8):
                // pre-fix `&summary[..200]` panicked on multi-byte UTF-8
                // boundaries (an attacker-supplied incident summary with
                // emoji at the right offset DoSed the KG ingest path).
                let trunc = crate::text_util::safe_truncate(summary, 200);
                edge.properties
                    .insert(intern("summary"), serde_json::Value::from(trunc));
            }
        }
        if let Some(ref sev) = self._current_event_severity {
            if !edge.properties.contains_key("severity") {
                edge.properties
                    .insert(intern("severity"), serde_json::Value::from(sev.as_str()));
            }
        }

        let idx = self.edges.len();
        self.outgoing.entry(edge.from).or_default().push(idx);
        self.incoming.entry(edge.to).or_default().push(idx);
        self.memory_estimate += Self::estimate_edge_size(&edge);

        // Maintain `last_edge_ts` index for both endpoints. This replaces the
        // O(E_avg log E_avg) per-node `all_edges()` scan that
        // `enforce_memory_limit` used to do under memory pressure (the worst
        // possible time to allocate). See RECURRING_BUGS.md
        // "enforce_memory_limit allocates O(N × E) under memory pressure".
        let edge_from = edge.from;
        let edge_to = edge.to;
        let edge_ts = edge.ts;
        self.last_edge_ts
            .entry(edge_from)
            .and_modify(|t| {
                if edge_ts > *t {
                    *t = edge_ts;
                }
            })
            .or_insert(edge_ts);
        if edge_to != edge_from {
            self.last_edge_ts
                .entry(edge_to)
                .and_modify(|t| {
                    if edge_ts > *t {
                        *t = edge_ts;
                    }
                })
                .or_insert(edge_ts);
        }

        self.edges.push(edge);

        // Real-time critical triggers (O(1) checks per edge)
        // Clone the minimal data needed to avoid borrow conflict
        let trigger_rel = self.edges[idx].relation;
        let trigger_from = self.edges[idx].from;
        let trigger_to = self.edges[idx].to;
        let trigger_ts = self.edges[idx].ts;
        let trigger_props = self.edges[idx].properties.clone();
        if !self.edges[idx].is_snapshot() {
            super::triggers::check_critical_triggers(
                self,
                trigger_rel,
                trigger_from,
                trigger_to,
                trigger_ts,
                &trigger_props,
            );
        }

        idx
    }

    #[allow(dead_code)] // part of KnowledgeGraph public surface; kept for future callers
    pub fn get_edge(&self, idx: usize) -> Option<&Edge> {
        self.edges.get(idx)
    }

    pub fn edge_count(&self) -> usize {
        self.edges.len()
    }

    pub fn node_count(&self) -> usize {
        self.nodes.len()
    }

    // ── Lookup by dedup key ─────────────────────────────────────────────

    pub fn find_by_pid(&self, pid: u32) -> Option<NodeId> {
        self.pid_index.get(&pid).copied()
    }

    pub fn find_by_ip(&self, addr: &str) -> Option<NodeId> {
        self.ip_index.get(addr).copied()
    }

    pub fn find_by_path(&self, path: &str) -> Option<NodeId> {
        self.file_index.get(path).copied()
    }

    pub fn find_by_user(&self, name: &str) -> Option<NodeId> {
        self.user_index.get(name).copied()
    }

    pub fn find_by_domain(&self, name: &str) -> Option<NodeId> {
        self.domain_index.get(name).copied()
    }

    pub fn find_by_port(&self, number: u16, protocol: &str) -> Option<NodeId> {
        self.port_index
            .get(&(number, protocol.to_string()))
            .copied()
    }

    pub fn find_by_container(&self, id: &str) -> Option<NodeId> {
        self.container_index.get(id).copied()
    }

    pub fn find_by_incident(&self, incident_id: &str) -> Option<NodeId> {
        self.incident_index.get(incident_id).copied()
    }

    #[allow(dead_code)] // campaign lookup kept for spec 016 campaign tab wiring
    pub fn find_by_campaign(&self, campaign_id: &str) -> Option<NodeId> {
        self.campaign_index.get(campaign_id).copied()
    }

    pub fn system_node(&self) -> Option<NodeId> {
        self.system_node
    }

    /// Phase 7 Gap 1: mark an Incident node as false positive.
    pub fn mark_false_positive(&mut self, node_id: NodeId, reporter: &str) {
        if let Some(Node::Incident {
            false_positive,
            fp_reporter,
            fp_reported_at,
            ..
        }) = self.nodes.get_mut(&node_id)
        {
            *false_positive = true;
            *fp_reporter = Some(reporter.to_string());
            *fp_reported_at = Some(chrono::Utc::now());
        }
    }

    /// Phase 014-C: Link incidents in a correlation chain via CorrelatedWith edges.
    /// Given a list of incident IDs that form a multi-stage attack chain,
    /// connects each pair with a CorrelatedWith edge for graph traversal.
    #[allow(dead_code)] // entity-overlap path wires chains today; kept for explicit chain-id usage
    pub fn link_correlated_incidents(&mut self, incident_ids: &[String], chain_id: &str) {
        let node_ids: Vec<NodeId> = incident_ids
            .iter()
            .filter_map(|iid| self.find_by_incident(iid))
            .collect();
        if node_ids.len() < 2 {
            return;
        }
        let now = chrono::Utc::now();
        // Create a complete graph of CorrelatedWith edges between all pairs
        for i in 0..node_ids.len() {
            for j in (i + 1)..node_ids.len() {
                let edge = Edge::new(node_ids[i], node_ids[j], Relation::CorrelatedWith, now)
                    .with_prop("chain_id", serde_json::Value::from(chain_id));
                self.add_edge(edge);
            }
        }
    }

    // ── Ensure helpers (find or create) ─────────────────────────────────

    pub fn ensure_process(
        &mut self,
        pid: u32,
        ppid: u32,
        comm: &str,
        uid: u32,
        ts: DateTime<Utc>,
    ) -> NodeId {
        if let Some(id) = self.find_by_pid(pid) {
            // Backfill missing fields from later events that carry richer context.
            // ensure_process is called from many event types (exec, exit, openat,
            // ptrace, etc) — earlier ones may lack comm/ppid/uid. Without backfill,
            // 70% of process nodes end up with empty comm in production.
            if let Some(Node::Process {
                comm: existing_comm,
                ppid: existing_ppid,
                uid: existing_uid,
                ..
            }) = self.nodes.get_mut(&id)
            {
                if existing_comm.is_empty() && !comm.is_empty() {
                    *existing_comm = comm.to_string();
                }
                if *existing_ppid == 0 && ppid > 0 {
                    *existing_ppid = ppid;
                }
                if *existing_uid == 0 && uid > 0 {
                    *existing_uid = uid;
                }
            }
            return id;
        }
        self.add_node(Node::Process {
            pid,
            ppid,
            comm: comm.to_string(),
            exe: None,
            uid,
            container_id: None,
            start_ts: ts,
            exit_ts: None,
        })
    }

    pub fn ensure_ip(&mut self, addr: &str, ts: DateTime<Utc>) -> NodeId {
        if let Some(id) = self.find_by_ip(addr) {
            // Update last_seen
            if let Some(Node::Ip { last_seen, .. }) = self.nodes.get_mut(&id) {
                *last_seen = ts;
            }
            return id;
        }
        self.add_node(Node::Ip {
            addr: addr.to_string(),
            is_internal: is_internal_ip(addr),
            datasets: Vec::new(),
            risk_score: 0,
            is_tor: false,
            first_seen: ts,
            last_seen: ts,
            attempted_usernames: Vec::new(),
        })
    }

    /// Spec 015: record an attacker-supplied username from a failed SSH auth.
    /// Appends to the Ip node's `attempted_usernames` list, deduplicating and
    /// capping at 50 most-recent entries (LIFO). Silently no-ops if the node
    /// is not an Ip.
    pub fn record_attempted_username(&mut self, ip_id: NodeId, attacker_login: &str) {
        const MAX_ATTEMPTED: usize = 50;
        if attacker_login.is_empty() {
            return;
        }
        if let Some(Node::Ip {
            attempted_usernames,
            ..
        }) = self.nodes.get_mut(&ip_id)
        {
            // lgtm[rust/cleartext-logging] — these are attacker-supplied brute-force
            // login names (e.g. "root", "admin"), NOT real credentials.
            // Stored for threat DNA fingerprinting.
            if let Some(pos) = attempted_usernames.iter().position(|u| u == attacker_login) {
                attempted_usernames.remove(pos);
            }
            attempted_usernames.push(attacker_login.to_string());
            // Cap (LIFO): drop the oldest entries when exceeding the max.
            if attempted_usernames.len() > MAX_ATTEMPTED {
                let drop = attempted_usernames.len() - MAX_ATTEMPTED;
                attempted_usernames.drain(0..drop);
            }
        }
    }

    /// Spec 015: retrieve attacker-supplied usernames from a given Ip node,
    /// for attacker fingerprinting / threat-dna features.
    #[allow(dead_code)] // API surface for future attacker-intel consumers
    pub fn attempted_usernames_for_ip(&self, ip_id: NodeId) -> Vec<String> {
        match self.nodes.get(&ip_id) {
            Some(Node::Ip {
                attempted_usernames,
                ..
            }) => attempted_usernames.clone(),
            _ => Vec::new(),
        }
    }

    pub fn ensure_file(&mut self, path: &str) -> NodeId {
        if let Some(id) = self.find_by_path(path) {
            return id;
        }
        self.add_node(Node::File {
            path: path.to_string(),
            sha256: None,
            size: None,
            entropy: None,
            is_sensitive: is_sensitive_path(path),
            yara_matches: Vec::new(),
        })
    }

    pub fn ensure_user(&mut self, name: &str) -> NodeId {
        if let Some(id) = self.find_by_user(name) {
            return id;
        }
        self.add_node(Node::User {
            name: name.to_string(),
            uid: None,
        })
    }

    pub fn ensure_domain(&mut self, name: &str) -> NodeId {
        if let Some(id) = self.find_by_domain(name) {
            return id;
        }
        self.add_node(Node::Domain {
            name: name.to_string(),
            datasets: Vec::new(),
            is_dga: None,
            entropy: None,
        })
    }

    pub fn ensure_port(&mut self, number: u16, protocol: &str) -> NodeId {
        if let Some(id) = self.find_by_port(number, protocol) {
            return id;
        }
        self.add_node(Node::Port {
            number,
            protocol: protocol.to_string(),
        })
    }

    pub fn ensure_container(&mut self, container_id: &str) -> NodeId {
        if let Some(id) = self.find_by_container(container_id) {
            return id;
        }
        self.add_node(Node::Container {
            container_id: container_id.to_string(),
            name: None,
            image: None,
            start_ts: None,
            exit_ts: None,
            oom_killed: false,
        })
    }

    /// Return the singleton System node id, creating it if necessary.
    ///
    /// The first caller may not know the real hostname (e.g. the decision
    /// recording path calls `ensure_system("")` before any event has flowed
    /// through). In that case the node is created with an empty label, and
    /// later calls that *do* have a real hostname will upgrade the label in
    /// place. Once a non-empty hostname is recorded it becomes immutable —
    /// the graph is single-host so there's no legitimate reason to rename.
    pub fn ensure_system(&mut self, hostname: &str) -> NodeId {
        if let Some(id) = self.system_node {
            // Late-upgrade: fill an empty hostname the first time a real one
            // shows up. This fixes the bug where the System node was stuck
            // with "" because the decision path happened to create it before
            // any ingested event did.
            if !hostname.is_empty() {
                if let Some(Node::System { hostname: h, .. }) = self.get_node_mut(id) {
                    if h.is_empty() {
                        *h = hostname.to_string();
                    }
                }
            }
            return id;
        }
        let id = self.add_node(Node::System {
            hostname: hostname.to_string(),
            sysctl_params: HashMap::new(),
        });
        self.system_node = Some(id);
        id
    }

    // ── Traversal primitives ────────────────────────────────────────────

    pub fn outgoing_edges(&self, node: NodeId) -> Vec<&Edge> {
        self.outgoing
            .get(&node)
            .map(|idxs| idxs.iter().filter_map(|&i| self.edges.get(i)).collect())
            .unwrap_or_default()
    }

    pub fn incoming_edges(&self, node: NodeId) -> Vec<&Edge> {
        self.incoming
            .get(&node)
            .map(|idxs| idxs.iter().filter_map(|&i| self.edges.get(i)).collect())
            .unwrap_or_default()
    }

    /// All edges (in + out) for a node, sorted by timestamp. Deduplicated by index.
    pub fn all_edges(&self, node: NodeId) -> Vec<&Edge> {
        let mut seen = HashSet::new();
        let out = self
            .outgoing
            .get(&node)
            .map(|v| v.as_slice())
            .unwrap_or(&[]);
        let inc = self
            .incoming
            .get(&node)
            .map(|v| v.as_slice())
            .unwrap_or(&[]);

        let mut edges: Vec<&Edge> = out
            .iter()
            .chain(inc.iter())
            .filter(|&&idx| seen.insert(idx))
            .filter_map(|&idx| self.edges.get(idx))
            .collect();
        edges.sort_by_key(|e| e.ts);
        edges
    }

    /// Find all nodes of a given type.
    pub fn nodes_of_type(&self, nt: NodeType) -> Vec<NodeId> {
        self.nodes
            .iter()
            .filter(|(_, n)| n.node_type() == nt)
            .map(|(&id, _)| id)
            .collect()
    }

    /// Find nodes that have edges newer than `since`. Much faster than scanning all nodes.
    pub fn active_nodes_since(&self, since: DateTime<Utc>) -> HashSet<NodeId> {
        let mut active = HashSet::new();
        // Scan edges from the end (most recent) backwards
        for edge in self.edges.iter().rev() {
            if edge.ts < since {
                break; // Edges are appended chronologically
            }
            active.insert(edge.from);
            active.insert(edge.to);
        }
        active
    }

    /// BFS neighborhood: subgraph within `depth` hops of `start`.
    pub fn neighborhood(&self, start: NodeId, depth: usize) -> SubGraph {
        // Caps prevent the BFS from exploding on hot nodes (a single attacker
        // IP can have thousands of incident edges). The frontend cytoscape viz
        // is unusable beyond a few hundred nodes anyway, and /api/graph/neighborhood
        // was the reason the dashboard hung on Investigation pivots.
        const MAX_TOTAL_EDGES: usize = 500;
        const MAX_EDGES_PER_NODE: usize = 50;

        let mut sub = SubGraph::default();
        let mut visited: HashSet<NodeId> = HashSet::new();
        let mut queue: VecDeque<(NodeId, usize)> = VecDeque::new();

        if let Some(node) = self.nodes.get(&start) {
            visited.insert(start);
            sub.nodes.insert(start, node.clone());
            queue.push_back((start, 0));
        }

        'bfs: while let Some((current, d)) = queue.pop_front() {
            if d >= depth {
                continue;
            }

            let mut per_node_edges = 0usize;

            // Outgoing (capped per-node)
            for edge in self.outgoing_edges(current) {
                if sub.edges.len() >= MAX_TOTAL_EDGES {
                    break 'bfs;
                }
                if per_node_edges >= MAX_EDGES_PER_NODE {
                    break;
                }
                sub.edges.push(edge.clone());
                per_node_edges += 1;
                if !visited.contains(&edge.to) {
                    visited.insert(edge.to);
                    if let Some(n) = self.nodes.get(&edge.to) {
                        sub.nodes.insert(edge.to, n.clone());
                    }
                    queue.push_back((edge.to, d + 1));
                }
            }

            // Incoming (capped per-node, shares budget with outgoing)
            for edge in self.incoming_edges(current) {
                if sub.edges.len() >= MAX_TOTAL_EDGES {
                    break 'bfs;
                }
                if per_node_edges >= MAX_EDGES_PER_NODE * 2 {
                    break;
                }
                sub.edges.push(edge.clone());
                per_node_edges += 1;
                if !visited.contains(&edge.from) {
                    visited.insert(edge.from);
                    if let Some(n) = self.nodes.get(&edge.from) {
                        sub.nodes.insert(edge.from, n.clone());
                    }
                    queue.push_back((edge.from, d + 1));
                }
            }
        }

        sub
    }

    /// BFS shortest path between two nodes. Returns edge chain or None.
    pub fn path_between(&self, from: NodeId, to: NodeId, max_depth: usize) -> Option<Vec<Edge>> {
        if from == to {
            return Some(Vec::new());
        }

        let mut visited: HashSet<NodeId> = HashSet::new();
        // (current_node, path_of_edge_indexes)
        let mut queue: VecDeque<(NodeId, Vec<usize>)> = VecDeque::new();

        visited.insert(from);
        queue.push_back((from, Vec::new()));

        while let Some((current, path)) = queue.pop_front() {
            if path.len() >= max_depth {
                continue;
            }

            // Outgoing edges
            if let Some(idxs) = self.outgoing.get(&current) {
                for &idx in idxs {
                    if let Some(edge) = self.edges.get(idx) {
                        if edge.to == to {
                            let mut result: Vec<Edge> = path
                                .iter()
                                .filter_map(|&i| self.edges.get(i).cloned())
                                .collect();
                            result.push(edge.clone());
                            return Some(result);
                        }
                        if !visited.contains(&edge.to) {
                            visited.insert(edge.to);
                            let mut new_path = path.clone();
                            new_path.push(idx);
                            queue.push_back((edge.to, new_path));
                        }
                    }
                }
            }

            // Incoming edges (bidirectional search)
            if let Some(idxs) = self.incoming.get(&current) {
                for &idx in idxs {
                    if let Some(edge) = self.edges.get(idx) {
                        if edge.from == to {
                            let mut result: Vec<Edge> = path
                                .iter()
                                .filter_map(|&i| self.edges.get(i).cloned())
                                .collect();
                            result.push(edge.clone());
                            return Some(result);
                        }
                        if !visited.contains(&edge.from) {
                            visited.insert(edge.from);
                            let mut new_path = path.clone();
                            new_path.push(idx);
                            queue.push_back((edge.from, new_path));
                        }
                    }
                }
            }
        }

        None
    }

    /// Process tree: all descendants via SpawnedBy edges (reversed).
    pub fn descendants(&self, pid: u32) -> Vec<NodeId> {
        let start = match self.find_by_pid(pid) {
            Some(id) => id,
            None => return Vec::new(),
        };

        let mut result = Vec::new();
        let mut queue = VecDeque::new();
        let mut visited = HashSet::new();

        queue.push_back(start);
        visited.insert(start);

        while let Some(current) = queue.pop_front() {
            // Children are nodes that have SpawnedBy edge pointing TO current
            if let Some(idxs) = self.incoming.get(&current) {
                for &idx in idxs {
                    if let Some(edge) = self.edges.get(idx) {
                        if edge.relation == Relation::SpawnedBy && !visited.contains(&edge.from) {
                            visited.insert(edge.from);
                            result.push(edge.from);
                            queue.push_back(edge.from);
                        }
                    }
                }
            }
        }

        result
    }

    /// Process tree: ancestors via SpawnedBy edges (forward).
    pub fn ancestors(&self, pid: u32) -> Vec<NodeId> {
        let start = match self.find_by_pid(pid) {
            Some(id) => id,
            None => return Vec::new(),
        };

        let mut result = Vec::new();
        let mut current = start;
        let mut visited = HashSet::new();
        visited.insert(current);

        loop {
            // Parent is the node this process has a SpawnedBy edge TO
            let parent = self.outgoing.get(&current).and_then(|idxs| {
                idxs.iter().find_map(|&idx| {
                    self.edges.get(idx).and_then(|e| {
                        if e.relation == Relation::SpawnedBy {
                            Some(e.to)
                        } else {
                            None
                        }
                    })
                })
            });

            match parent {
                Some(p) if !visited.contains(&p) => {
                    visited.insert(p);
                    result.push(p);
                    current = p;
                }
                _ => break,
            }
        }

        result
    }

    /// Timeline: all edges of a node sorted by timestamp, excluding snapshots.
    pub fn timeline(&self, node: NodeId) -> Vec<&Edge> {
        let mut edges = self.all_edges(node);
        edges.retain(|e| !e.is_snapshot());
        edges
    }

    /// Threat intel hits: processes connected to IPs with non-empty datasets.
    pub fn threat_intel_hits(&self) -> Vec<(NodeId, NodeId, String)> {
        let mut hits = Vec::new();
        for &ip_id in &self.threat_intel_nodes {
            let datasets = match self.nodes.get(&ip_id) {
                Some(Node::Ip { datasets, .. }) if !datasets.is_empty() => datasets.clone(),
                _ => continue,
            };

            // Find processes connected to this IP
            if let Some(idxs) = self.incoming.get(&ip_id) {
                for &idx in idxs {
                    if let Some(edge) = self.edges.get(idx) {
                        if edge.relation == Relation::ConnectedTo {
                            for ds in &datasets {
                                hits.push((edge.from, ip_id, ds.clone()));
                            }
                        }
                    }
                }
            }
        }
        hits
    }

    /// Nodes matching a type and predicate.
    #[allow(dead_code)] // general query helper kept on KG public surface
    pub fn find_nodes(
        &self,
        node_type: NodeType,
        predicate: impl Fn(&Node) -> bool,
    ) -> Vec<NodeId> {
        self.nodes
            .iter()
            .filter(|(_, n)| n.node_type() == node_type && predicate(n))
            .map(|(&id, _)| id)
            .collect()
    }

    // ── TTL + Memory management ─────────────────────────────────────────

    /// Remove expired nodes based on TTL rules.
    pub fn cleanup_expired(&mut self, now: DateTime<Utc>) {
        let expired: Vec<NodeId> = self
            .nodes
            .iter()
            .filter(|(id, node)| self.is_expired(node, **id, now))
            .map(|(&id, _)| id)
            .collect();

        for id in expired {
            self.remove_node(id);
        }
    }

    /// If memory exceeds max, prune oldest nodes by last edge timestamp (LRU).
    ///
    /// Uses the `last_edge_ts` index (maintained on every `add_edge`) to
    /// resolve each node's last activity in O(1). The previous implementation
    /// called `all_edges(id)` per node, which allocates a `Vec<&Edge>` and
    /// sort-merges in/out adjacency lists — O(N × E_avg log E_avg) in total,
    /// and the worst possible time to allocate is exactly when memory
    /// pressure has triggered this path. See `RECURRING_BUGS.md`
    /// "enforce_memory_limit allocates O(N × E) under memory pressure".
    pub fn enforce_memory_limit(&mut self) {
        if self.memory_estimate <= self.max_memory {
            return;
        }

        // Collect (node_id, last_edge_ts) sorted oldest first.
        // Nodes with no edges fall back to `created_at` so they are not
        // unfairly evicted ahead of the working set.
        let mut candidates: Vec<(NodeId, DateTime<Utc>)> = self
            .nodes
            .iter()
            .filter(|(_, node)| !matches!(node, Node::System { .. }))
            .map(|(&id, _)| {
                let last_ts = self
                    .last_edge_ts
                    .get(&id)
                    .copied()
                    .unwrap_or(self.created_at);
                (id, last_ts)
            })
            .collect();

        candidates.sort_by_key(|(_, ts)| *ts);

        for (id, _) in candidates {
            if self.memory_estimate <= self.max_memory {
                break;
            }
            self.remove_node(id);
        }
    }

    /// Current estimated memory usage in bytes.
    #[allow(dead_code)] // exposed via metrics() but also retained for direct callers
    pub fn memory_estimate(&self) -> usize {
        self.memory_estimate
    }

    // ── Metrics ─────────────────────────────────────────────────────────

    pub fn metrics(&self) -> GraphMetrics {
        let mut nodes_by_type: HashMap<String, usize> = HashMap::new();
        for node in self.nodes.values() {
            *nodes_by_type
                .entry(format!("{:?}", node.node_type()))
                .or_default() += 1;
        }

        // Total degree from adjacency-list lengths. Pre-2026-04-23 this
        // was `self.nodes.keys().map(|id| self.all_edges(*id).len()).sum()`
        // which allocated + sorted a `Vec<&Edge>` per node — O(N × E_avg
        // log E_avg) and called from inside the slow_loop write lock
        // that also held disk + SQLite I/O. Same anti-pattern
        // `enforce_memory_limit` had; this is its sibling. Self-loops
        // contribute +2 here vs +1 in the old all_edges (which de-duped
        // the same edge appearing in both outgoing[A] and incoming[A]);
        // we accept the rounding — avg_degree is diagnostic only.
        let outgoing_total: usize = self.outgoing.values().map(Vec::len).sum();
        let incoming_total: usize = self.incoming.values().map(Vec::len).sum();
        let total_degree = outgoing_total + incoming_total;
        let avg_degree = if self.nodes.is_empty() {
            0.0
        } else {
            total_degree as f32 / self.nodes.len() as f32
        };

        GraphMetrics {
            node_count: self.nodes.len(),
            edge_count: self.edges.len(),
            memory_bytes: self.memory_estimate,
            nodes_by_type,
            avg_degree,
            threat_intel_nodes: self.threat_intel_nodes.len(),
            incident_nodes: self.incident_index.len(),
        }
    }

    // ── Serialization support ───────────────────────────────────────────

    pub fn nodes(&self) -> &HashMap<NodeId, Node> {
        &self.nodes
    }

    pub fn edges_slice(&self) -> &[Edge] {
        &self.edges
    }

    // ── Aggregation helpers (Phase 3B) ───────────────────────────────────

    /// Count outgoing edges of a specific relation type within a time window.
    pub fn count_edges_in_window(
        &self,
        node: NodeId,
        relation: Relation,
        now: chrono::DateTime<chrono::Utc>,
        window: chrono::Duration,
    ) -> usize {
        self.outgoing_edges(node)
            .iter()
            .filter(|e| e.relation == relation && now - e.ts < window)
            .count()
    }

    /// Count distinct destination nodes connected via a relation in a time window.
    pub fn count_distinct_targets_in_window(
        &self,
        node: NodeId,
        relation: Relation,
        now: chrono::DateTime<chrono::Utc>,
        window: chrono::Duration,
    ) -> usize {
        let targets: std::collections::HashSet<NodeId> = self
            .outgoing_edges(node)
            .iter()
            .filter(|e| e.relation == relation && now - e.ts < window)
            .map(|e| e.to)
            .collect();
        targets.len()
    }

    /// Collect all outgoing edges of a relation within a window, returning (target_id, edge_ref).
    pub fn edges_in_window(
        &self,
        node: NodeId,
        relation: Relation,
        now: chrono::DateTime<chrono::Utc>,
        window: chrono::Duration,
    ) -> Vec<&Edge> {
        self.outgoing_edges(node)
            .into_iter()
            .filter(|e| e.relation == relation && now - e.ts < window)
            .collect()
    }

    // ── Internal helpers ────────────────────────────────────────────────

    fn find_existing(&self, node: &Node) -> Option<NodeId> {
        match node {
            Node::Process { pid, .. } => self.pid_index.get(pid).copied(),
            Node::Ip { addr, .. } => self.ip_index.get(addr).copied(),
            Node::File { path, .. } => self.file_index.get(path).copied(),
            Node::User { name, .. } => self.user_index.get(name).copied(),
            Node::Domain { name, .. } => self.domain_index.get(name).copied(),
            Node::Port {
                number, protocol, ..
            } => self.port_index.get(&(*number, protocol.clone())).copied(),
            Node::Container { container_id, .. } => self.container_index.get(container_id).copied(),
            Node::Device {
                vendor,
                product,
                serial,
                ..
            } => {
                let key = device_key(vendor, product, serial.as_deref());
                self.device_index.get(&key).copied()
            }
            Node::System { .. } => self.system_node,
            Node::Incident { incident_id, .. } => self.incident_index.get(incident_id).copied(),
            Node::Campaign { campaign_id, .. } => self.campaign_index.get(campaign_id).copied(),
        }
    }

    pub(crate) fn index_node(&mut self, id: NodeId, node: &Node) {
        match node {
            Node::Process { pid, .. } => {
                self.pid_index.insert(*pid, id);
            }
            Node::Ip { addr, datasets, .. } => {
                self.ip_index.insert(addr.clone(), id);
                if !datasets.is_empty() {
                    self.threat_intel_nodes.insert(id);
                }
            }
            Node::File { path, .. } => {
                self.file_index.insert(path.clone(), id);
            }
            Node::User { name, .. } => {
                self.user_index.insert(name.clone(), id);
            }
            Node::Domain { name, .. } => {
                self.domain_index.insert(name.clone(), id);
            }
            Node::Port {
                number, protocol, ..
            } => {
                self.port_index.insert((*number, protocol.clone()), id);
            }
            Node::Container { container_id, .. } => {
                self.container_index.insert(container_id.clone(), id);
            }
            Node::Device {
                vendor,
                product,
                serial,
                ..
            } => {
                let key = device_key(vendor, product, serial.as_deref());
                self.device_index.insert(key, id);
            }
            Node::System { .. } => {
                self.system_node = Some(id);
            }
            Node::Incident { incident_id, .. } => {
                self.incident_index.insert(incident_id.clone(), id);
            }
            Node::Campaign { campaign_id, .. } => {
                self.campaign_index.insert(campaign_id.clone(), id);
            }
        }
    }

    fn deindex_node(&mut self, id: NodeId, node: &Node) {
        match node {
            Node::Process { pid, .. } => {
                self.pid_index.remove(pid);
            }
            Node::Ip { addr, .. } => {
                self.ip_index.remove(addr);
            }
            Node::File { path, .. } => {
                self.file_index.remove(path);
            }
            Node::User { name, .. } => {
                self.user_index.remove(name);
            }
            Node::Domain { name, .. } => {
                self.domain_index.remove(name);
            }
            Node::Port {
                number, protocol, ..
            } => {
                self.port_index.remove(&(*number, protocol.clone()));
            }
            Node::Container { container_id, .. } => {
                self.container_index.remove(container_id);
            }
            Node::Device {
                vendor,
                product,
                serial,
                ..
            } => {
                let key = device_key(vendor, product, serial.as_deref());
                self.device_index.remove(&key);
            }
            Node::System { .. } => {
                if self.system_node == Some(id) {
                    self.system_node = None;
                }
            }
            Node::Incident { incident_id, .. } => {
                self.incident_index.remove(incident_id);
            }
            Node::Campaign { campaign_id, .. } => {
                self.campaign_index.remove(campaign_id);
            }
        }
    }

    fn merge_node(existing: &mut Node, incoming: &Node) {
        match (existing, incoming) {
            (
                Node::Process {
                    exe,
                    exit_ts,
                    container_id,
                    ..
                },
                Node::Process {
                    exe: new_exe,
                    exit_ts: new_exit,
                    container_id: new_cid,
                    ..
                },
            ) => {
                if exe.is_none() && new_exe.is_some() {
                    *exe = new_exe.clone();
                }
                if exit_ts.is_none() && new_exit.is_some() {
                    *exit_ts = *new_exit;
                }
                if container_id.is_none() && new_cid.is_some() {
                    *container_id = new_cid.clone();
                }
            }
            (
                Node::Ip {
                    last_seen,
                    datasets,
                    risk_score,
                    is_tor,
                    ..
                },
                Node::Ip {
                    last_seen: new_last,
                    datasets: new_ds,
                    risk_score: new_rs,
                    is_tor: new_tor,
                    ..
                },
            ) => {
                if *new_last > *last_seen {
                    *last_seen = *new_last;
                }
                for ds in new_ds {
                    if !datasets.contains(ds) {
                        datasets.push(ds.clone());
                    }
                }
                if *new_rs > *risk_score {
                    *risk_score = *new_rs;
                }
                if *new_tor {
                    *is_tor = true;
                }
            }
            (
                Node::File {
                    sha256,
                    size,
                    entropy,
                    yara_matches,
                    ..
                },
                Node::File {
                    sha256: new_sha,
                    size: new_size,
                    entropy: new_ent,
                    yara_matches: new_yara,
                    ..
                },
            ) => {
                if sha256.is_none() && new_sha.is_some() {
                    *sha256 = new_sha.clone();
                }
                if size.is_none() && new_size.is_some() {
                    *size = *new_size;
                }
                if entropy.is_none() && new_ent.is_some() {
                    *entropy = *new_ent;
                }
                for y in new_yara {
                    if !yara_matches.contains(y) {
                        yara_matches.push(y.clone());
                    }
                }
            }
            (
                Node::Container {
                    name,
                    image,
                    start_ts,
                    exit_ts,
                    oom_killed,
                    ..
                },
                Node::Container {
                    name: nn,
                    image: ni,
                    start_ts: ns,
                    exit_ts: ne,
                    oom_killed: no,
                    ..
                },
            ) => {
                if name.is_none() && nn.is_some() {
                    *name = nn.clone();
                }
                if image.is_none() && ni.is_some() {
                    *image = ni.clone();
                }
                if start_ts.is_none() && ns.is_some() {
                    *start_ts = *ns;
                }
                if exit_ts.is_none() && ne.is_some() {
                    *exit_ts = *ne;
                }
                if *no {
                    *oom_killed = true;
                }
            }
            (
                Node::Domain {
                    datasets,
                    is_dga,
                    entropy,
                    ..
                },
                Node::Domain {
                    datasets: nd,
                    is_dga: ndga,
                    entropy: ne,
                    ..
                },
            ) => {
                for d in nd {
                    if !datasets.contains(d) {
                        datasets.push(d.clone());
                    }
                }
                if is_dga.is_none() && ndga.is_some() {
                    *is_dga = *ndga;
                }
                if entropy.is_none() && ne.is_some() {
                    *entropy = *ne;
                }
            }
            _ => {}
        }
    }

    fn is_expired(&self, node: &Node, node_id: NodeId, now: DateTime<Utc>) -> bool {
        match node {
            Node::Process {
                exit_ts: Some(exit),
                ..
            } => now - *exit > Duration::hours(1),
            Node::Process { exit_ts: None, .. } => {
                // Check last edge
                let last = self.all_edges(node_id).last().map(|e| e.ts);
                match last {
                    Some(ts) => now - ts > Duration::hours(24),
                    None => now - self.created_at > Duration::hours(24),
                }
            }
            Node::Ip {
                datasets,
                risk_score,
                last_seen,
                ..
            } => {
                if !datasets.is_empty() || *risk_score > 0 {
                    return false; // Permanent if threat intel
                }
                now - *last_seen > Duration::hours(24)
            }
            Node::File { is_sensitive, .. } => {
                if *is_sensitive {
                    return false; // Permanent
                }
                let last = self.all_edges(node_id).last().map(|e| e.ts);
                match last {
                    Some(ts) => now - ts > Duration::hours(24),
                    None => false,
                }
            }
            Node::User { .. } | Node::Device { .. } | Node::System { .. } => false, // Permanent
            Node::Domain { datasets, .. } => {
                if !datasets.is_empty() {
                    return false;
                }
                let last = self.all_edges(node_id).last().map(|e| e.ts);
                match last {
                    Some(ts) => now - ts > Duration::hours(24),
                    None => false,
                }
            }
            Node::Port { .. } => {
                let last = self.all_edges(node_id).last().map(|e| e.ts);
                match last {
                    Some(ts) => now - ts > Duration::hours(24),
                    None => false,
                }
            }
            Node::Container {
                exit_ts: Some(exit),
                ..
            } => now - *exit > Duration::hours(1),
            Node::Container { .. } => false, // Running
            Node::Incident { decision, ts, .. } => {
                // Keep active response incidents permanently for audit trail
                if matches!(
                    decision.as_deref(),
                    Some("block_ip")
                        | Some("kill_process")
                        | Some("suspend_user_sudo")
                        | Some("block_container")
                ) {
                    return false;
                }
                now - *ts > Duration::days(7)
            }
            Node::Campaign { last_seen, .. } => now - *last_seen > Duration::days(30),
        }
    }

    pub(crate) fn estimate_node_size(node: &Node) -> usize {
        match node {
            Node::Process { .. } => 250,
            Node::Ip { .. } => 200,
            Node::File { .. } => 200,
            Node::User { .. } => 100,
            Node::Domain { .. } => 150,
            Node::Port { .. } => 50,
            Node::Container { .. } => 200,
            Node::Device { .. } => 150,
            Node::System { .. } => 500,
            Node::Incident { .. } => 300,
            Node::Campaign { .. } => 200,
        }
    }

    pub(crate) fn estimate_edge_size(edge: &Edge) -> usize {
        120 + edge.properties.len() * 40
    }
}

// ── Utility functions ───────────────────────────────────────────────────

fn is_internal_ip(addr: &str) -> bool {
    addr.starts_with("10.")
        || addr.starts_with("172.16.")
        || addr.starts_with("172.17.")
        || addr.starts_with("172.18.")
        || addr.starts_with("172.19.")
        || addr.starts_with("172.2")
        || addr.starts_with("172.30.")
        || addr.starts_with("172.31.")
        || addr.starts_with("192.168.")
        || addr == "127.0.0.1"
        || addr == "::1"
}

fn is_sensitive_path(path: &str) -> bool {
    let sensitive = [
        "/etc/passwd",
        "/etc/shadow",
        "/etc/sudoers",
        "/etc/ssh/",
        "authorized_keys",
        "/etc/cron",
        "/etc/systemd/",
        "/usr/lib/systemd/",
        "/var/spool/cron",
        ".ssh/id_",
        ".ssh/config",
        ".bashrc",
        ".bash_profile",
        ".profile",
    ];
    sensitive.iter().any(|s| path.contains(s))
}

fn device_key(vendor: &str, product: &str, serial: Option<&str>) -> String {
    format!("{}:{}:{}", vendor, product, serial.unwrap_or(""))
}

#[cfg(test)]
mod tests {
    use super::*;

    fn ts(secs: i64) -> DateTime<Utc> {
        DateTime::from_timestamp(1700000000 + secs, 0).unwrap()
    }

    #[test]
    fn test_add_node_dedup() {
        let mut g = KnowledgeGraph::new();
        let id1 = g.ensure_process(1234, 1, "bash", 0, ts(0));
        let id2 = g.ensure_process(1234, 1, "bash", 0, ts(1));
        assert_eq!(id1, id2);
        assert_eq!(g.node_count(), 1);
    }

    #[test]
    fn test_add_edge_no_dedup() {
        let mut g = KnowledgeGraph::new();
        let proc_id = g.ensure_process(1, 0, "bash", 0, ts(0));
        let ip_id = g.ensure_ip("1.2.3.4", ts(0));
        g.add_edge(Edge::new(proc_id, ip_id, Relation::ConnectedTo, ts(1)));
        g.add_edge(Edge::new(proc_id, ip_id, Relation::ConnectedTo, ts(2)));
        assert_eq!(g.edge_count(), 2);
    }

    #[test]
    fn test_find_by_index() {
        let mut g = KnowledgeGraph::new();
        g.ensure_process(42, 1, "wget", 0, ts(0));
        g.ensure_ip("10.0.0.1", ts(0));
        g.ensure_file("/etc/passwd");
        g.ensure_user("root");
        g.ensure_domain("evil.com");
        g.ensure_port(443, "tcp");

        assert!(g.find_by_pid(42).is_some());
        assert!(g.find_by_ip("10.0.0.1").is_some());
        assert!(g.find_by_path("/etc/passwd").is_some());
        assert!(g.find_by_user("root").is_some());
        assert!(g.find_by_domain("evil.com").is_some());
        assert!(g.find_by_port(443, "tcp").is_some());
        assert!(g.find_by_pid(999).is_none());
    }

    #[test]
    fn test_sensitive_file() {
        let mut g = KnowledgeGraph::new();
        let id = g.ensure_file("/etc/shadow");
        assert!(g.get_node(id).unwrap().is_sensitive_file());

        let id2 = g.ensure_file("/tmp/foo.txt");
        assert!(!g.get_node(id2).unwrap().is_sensitive_file());
    }

    #[test]
    fn test_internal_ip() {
        let mut g = KnowledgeGraph::new();
        let id = g.ensure_ip("192.168.1.1", ts(0));
        match g.get_node(id) {
            Some(Node::Ip { is_internal, .. }) => assert!(is_internal),
            _ => panic!("expected Ip node"),
        }
        let id2 = g.ensure_ip("8.8.8.8", ts(0));
        match g.get_node(id2) {
            Some(Node::Ip { is_internal, .. }) => assert!(!is_internal),
            _ => panic!("expected Ip node"),
        }
    }

    #[test]
    fn test_remove_node() {
        let mut g = KnowledgeGraph::new();
        let id = g.ensure_ip("1.2.3.4", ts(0));
        assert_eq!(g.node_count(), 1);
        g.remove_node(id);
        assert_eq!(g.node_count(), 0);
        assert!(g.find_by_ip("1.2.3.4").is_none());
    }

    #[test]
    fn test_descendants_and_ancestors() {
        let mut g = KnowledgeGraph::new();
        let sshd = g.ensure_process(800, 1, "sshd", 0, ts(0));
        let bash = g.ensure_process(1234, 800, "bash", 0, ts(1));
        let wget = g.ensure_process(1235, 1234, "wget", 0, ts(2));
        let payload = g.ensure_process(1236, 1234, "payload", 0, ts(3));

        g.add_edge(Edge::new(bash, sshd, Relation::SpawnedBy, ts(1)));
        g.add_edge(Edge::new(wget, bash, Relation::SpawnedBy, ts(2)));
        g.add_edge(Edge::new(payload, bash, Relation::SpawnedBy, ts(3)));

        let desc = g.descendants(800);
        assert_eq!(desc.len(), 3); // bash, wget, payload

        let anc = g.ancestors(1236);
        assert_eq!(anc.len(), 2); // bash, sshd
        assert_eq!(anc[0], bash);
        assert_eq!(anc[1], sshd);
    }

    #[test]
    fn test_path_between() {
        let mut g = KnowledgeGraph::new();
        let user = g.ensure_user("root");
        let ip1 = g.ensure_ip("185.1.1.1", ts(0));
        let proc1 = g.ensure_process(1234, 1, "bash", 0, ts(1));
        let ip2 = g.ensure_ip("93.1.1.1", ts(2));

        g.add_edge(Edge::new(user, ip1, Relation::LoggedInFrom, ts(0)));
        g.add_edge(Edge::new(proc1, user, Relation::RunAs, ts(1)));
        g.add_edge(Edge::new(proc1, ip2, Relation::ConnectedTo, ts(2)));

        let path = g.path_between(ip1, ip2, 5);
        assert!(path.is_some());
        assert!(path.unwrap().len() >= 2);
    }

    #[test]
    fn test_neighborhood() {
        let mut g = KnowledgeGraph::new();
        let proc1 = g.ensure_process(1, 0, "bash", 0, ts(0));
        let ip = g.ensure_ip("1.2.3.4", ts(0));
        let file = g.ensure_file("/tmp/x");
        g.add_edge(Edge::new(proc1, ip, Relation::ConnectedTo, ts(1)));
        g.add_edge(Edge::new(proc1, file, Relation::Wrote, ts(2)));

        let sub = g.neighborhood(proc1, 1);
        assert_eq!(sub.nodes.len(), 3);
        assert_eq!(sub.edges.len(), 2);
    }

    #[test]
    fn test_timeline() {
        let mut g = KnowledgeGraph::new();
        let proc1 = g.ensure_process(1, 0, "bash", 0, ts(0));
        let ip = g.ensure_ip("1.2.3.4", ts(0));
        let file = g.ensure_file("/tmp/x");

        g.add_edge(Edge::new(proc1, file, Relation::Wrote, ts(3)));
        g.add_edge(Edge::new(proc1, ip, Relation::ConnectedTo, ts(1)));
        g.add_edge(Edge::new(proc1, file, Relation::Read, ts(2)));

        let tl = g.timeline(proc1);
        assert_eq!(tl.len(), 3);
        assert!(tl[0].ts <= tl[1].ts);
        assert!(tl[1].ts <= tl[2].ts);
    }

    #[test]
    fn test_threat_intel_hits() {
        let mut g = KnowledgeGraph::new();
        let proc1 = g.ensure_process(1, 0, "wget", 0, ts(0));
        let ip = g.add_node(Node::Ip {
            addr: "93.1.1.1".into(),
            is_internal: false,
            datasets: vec!["sslbl".into()],
            risk_score: 80,
            is_tor: false,
            first_seen: ts(0),
            last_seen: ts(0),
            attempted_usernames: Vec::new(),
        });
        g.add_edge(Edge::new(proc1, ip, Relation::ConnectedTo, ts(1)));

        let hits = g.threat_intel_hits();
        assert_eq!(hits.len(), 1);
        assert_eq!(hits[0].2, "sslbl");
    }

    #[test]
    fn test_system_singleton() {
        let mut g = KnowledgeGraph::new();
        let id1 = g.ensure_system("prod-01");
        let id2 = g.ensure_system("prod-01");
        assert_eq!(id1, id2);
        assert_eq!(g.node_count(), 1);
    }

    #[test]
    fn test_system_hostname_late_upgrade() {
        // Regression: the decision recording path used to call
        // ensure_system("") before any event had flowed through, creating
        // the singleton with an empty hostname. Subsequent calls with real
        // hostnames did nothing, so downstream consumers (investigation
        // clusters, reports) ended up with host="".
        let mut g = KnowledgeGraph::new();
        let first = g.ensure_system("");
        let second = g.ensure_system("real-host-01");
        assert_eq!(first, second, "still the same singleton");
        let node = g.get_node(first).unwrap();
        if let Node::System { hostname, .. } = node {
            assert_eq!(
                hostname, "real-host-01",
                "empty hostname should be upgraded"
            );
        } else {
            panic!("expected System node");
        }
        // And once set, further calls with different hostnames don't clobber.
        g.ensure_system("different");
        if let Some(Node::System { hostname, .. }) = g.get_node(first) {
            assert_eq!(hostname, "real-host-01");
        }
    }

    #[test]
    fn test_metrics() {
        let mut g = KnowledgeGraph::new();
        g.ensure_process(1, 0, "bash", 0, ts(0));
        g.ensure_ip("1.2.3.4", ts(0));
        let m = g.metrics();
        assert_eq!(m.node_count, 2);
        assert_eq!(m.edge_count, 0);
    }

    // ── last_edge_ts / enforce_memory_limit regression suite ─────────
    //
    // Anchors for `RECURRING_BUGS.md` "enforce_memory_limit allocates
    // O(N × E) under memory pressure". Cached `last_edge_ts` MUST stay
    // consistent with the slow-but-correct `all_edges()` derivation, AND
    // the LRU eviction order MUST be identical to the previous behaviour.

    fn build_graph_with_edges(edges: &[(u32, &str, i64)]) -> KnowledgeGraph {
        let mut g = KnowledgeGraph::new();
        for (pid, ip, secs) in edges {
            let p = g.ensure_process(*pid, 0, "test", 0, ts(*secs));
            let ip_node = g.ensure_ip(ip, ts(*secs));
            g.add_edge(Edge::new(p, ip_node, Relation::ConnectedTo, ts(*secs)));
        }
        g
    }

    fn slow_last_edge_ts(g: &KnowledgeGraph, id: NodeId) -> Option<chrono::DateTime<chrono::Utc>> {
        // Reference implementation: derive the answer by scanning every edge
        // touching `id`. The cached `last_edge_ts` must agree for every node.
        g.all_edges(id).iter().map(|e| e.ts).max()
    }

    #[test]
    fn last_edge_ts_matches_full_scan_after_inserts() {
        let g = build_graph_with_edges(&[
            (101, "1.1.1.1", 0),
            (101, "1.1.1.1", 60),
            (102, "2.2.2.2", 30),
        ]);
        for &id in g.nodes.keys() {
            let cached = g.last_edge_ts.get(&id).copied();
            let derived = slow_last_edge_ts(&g, id);
            assert_eq!(
                cached, derived,
                "cached last_edge_ts must equal full-scan derivation for node {id:?}"
            );
        }
    }

    #[test]
    fn last_edge_ts_takes_max_not_last_inserted() {
        // Insert out of chronological order — cache must hold the MAX timestamp,
        // not the most recently inserted.
        let mut g = KnowledgeGraph::new();
        g.ensure_process(101, 0, "test", 0, ts(0));
        g.ensure_ip("1.1.1.1", ts(0));
        let p = g.find_by_pid(101).unwrap();
        let ip = g.find_by_ip("1.1.1.1").unwrap();
        g.add_edge(Edge::new(p, ip, Relation::ConnectedTo, ts(100)));
        g.add_edge(Edge::new(p, ip, Relation::ConnectedTo, ts(50))); // older
        g.add_edge(Edge::new(p, ip, Relation::ConnectedTo, ts(80))); // still older than 100

        assert_eq!(g.last_edge_ts.get(&p).copied(), Some(ts(100)));
        assert_eq!(g.last_edge_ts.get(&ip).copied(), Some(ts(100)));
    }

    #[test]
    fn last_edge_ts_cleared_on_remove_node() {
        let mut g = build_graph_with_edges(&[(303, "9.9.9.9", 10)]);
        let p = g.find_by_pid(303).unwrap();
        assert!(g.last_edge_ts.contains_key(&p));
        g.remove_node(p);
        assert!(
            !g.last_edge_ts.contains_key(&p),
            "removing a node must drop its last_edge_ts entry"
        );
    }

    #[test]
    fn enforce_memory_limit_evicts_oldest_first() {
        // Three IP nodes touched at t=0, t=100, t=200. With max_memory clamped
        // below the actual estimate, the t=0 node must be evicted before the
        // others (LRU eviction).
        let mut g = KnowledgeGraph::new();
        let p = g.ensure_process(7, 0, "p", 0, ts(0));
        let a = g.ensure_ip("10.0.0.1", ts(0));
        let b = g.ensure_ip("10.0.0.2", ts(100));
        let c = g.ensure_ip("10.0.0.3", ts(200));
        g.add_edge(Edge::new(p, a, Relation::ConnectedTo, ts(0)));
        g.add_edge(Edge::new(p, b, Relation::ConnectedTo, ts(100)));
        g.add_edge(Edge::new(p, c, Relation::ConnectedTo, ts(200)));

        // Force eviction by clamping the budget below the current estimate
        // but above what any single node + edge weighs, so we evict exactly one.
        g.max_memory = g.memory_estimate.saturating_sub(1);
        g.enforce_memory_limit();

        // Oldest IP should be gone, newer two retained.
        assert!(
            g.find_by_ip("10.0.0.1").is_none(),
            "oldest IP must be evicted"
        );
        assert!(g.find_by_ip("10.0.0.2").is_some());
        assert!(g.find_by_ip("10.0.0.3").is_some());
    }

    #[test]
    fn enforce_memory_limit_no_op_when_under_budget() {
        let mut g = build_graph_with_edges(&[(1, "1.1.1.1", 0)]);
        let nodes_before = g.nodes.len();
        g.max_memory = usize::MAX;
        g.enforce_memory_limit();
        assert_eq!(g.nodes.len(), nodes_before);
    }

    #[test]
    fn last_edge_ts_handles_self_loop_without_double_count() {
        // Self-loop: from == to. Inserting an edge from a node to itself must
        // not panic and must record the timestamp once. Defensive guard for
        // a class of callers we may add later (e.g. SystemEvent → System).
        let mut g = KnowledgeGraph::new();
        let p = g.ensure_process(1, 0, "p", 0, ts(0));
        g.add_edge(Edge::new(p, p, Relation::ConnectedTo, ts(42)));
        assert_eq!(g.last_edge_ts.get(&p).copied(), Some(ts(42)));
    }

    #[test]
    fn last_edge_ts_rebuilds_from_snapshot_load() {
        // The cache is NOT in the wire format. Loading an old snapshot must
        // rebuild it from the edge vector; otherwise enforce_memory_limit on a
        // freshly-restored agent would treat every node as "as old as graph
        // creation time" and evict things at random.
        use tempfile::tempdir;
        let dir = tempdir().unwrap();
        let path = dir.path().join("snap.json");

        let g_before = build_graph_with_edges(&[(1, "1.1.1.1", 0), (2, "2.2.2.2", 60)]);
        g_before.save_snapshot(&path).unwrap();
        let g_after = KnowledgeGraph::load_snapshot(&path);

        for (id_before, _) in g_before.nodes.iter() {
            // Find the corresponding node in g_after by entity, since IDs
            // may differ across reload.
            let derived_before = slow_last_edge_ts(&g_before, *id_before);
            let cached_before = g_before.last_edge_ts.get(id_before).copied();
            assert_eq!(derived_before, cached_before);
        }
        // Rebuilt cache must match a full scan on the loaded graph.
        for &id in g_after.nodes.keys() {
            let derived = slow_last_edge_ts(&g_after, id);
            let cached = g_after.last_edge_ts.get(&id).copied();
            assert_eq!(
                cached, derived,
                "loaded graph: cached last_edge_ts must match full-scan derivation"
            );
        }
    }

    /// 2026-05-02 KG memory phase 1: property keys on Edge are now
    /// interned `Arc<str>` so the same key shared across thousands of
    /// edges costs ONE allocation, not thousands. The anchor proves
    /// the dedup at the data structure level: two edges enriched by
    /// `add_edge` end up holding `Arc::ptr_eq` keys, and the global
    /// pool reports one entry for that key regardless of insertion
    /// volume.
    ///
    /// Regression risk: a future change that calls `properties.insert`
    /// with a fresh `Arc::from(&str)` instead of `intern(&str)` would
    /// break sharing without breaking unit-test correctness — only a
    /// pointer-equality anchor catches that drift.
    #[test]
    fn edge_property_keys_are_shared_via_interner() {
        use std::sync::Arc;

        // Two independent edges, each constructed via `with_prop`
        // which routes the key through the global interner. The
        // anchor proves the dedup at the data structure level: the
        // resulting `Arc<str>` keys are pointer-equal across edges.
        let now = Utc::now();
        let edge1 = Edge::new(1, 2, Relation::Executed, now)
            .with_prop("event_source", "ssh")
            .with_prop("event_kind", "auth.login");
        let edge2 = Edge::new(3, 4, Relation::Executed, now)
            .with_prop("event_source", "ssh")
            .with_prop("event_kind", "auth.login");

        let key1 = edge1
            .properties
            .keys()
            .find(|k| &***k == "event_source")
            .expect("event_source key present on first edge");
        let key2 = edge2
            .properties
            .keys()
            .find(|k| &***k == "event_source")
            .expect("event_source key present on second edge");

        // Pointer equality across edges, not just structural
        // equality. A future regression that calls `Arc::from(s)`
        // directly instead of `intern(s)` would still satisfy
        // `key1 == key2` but break this assertion.
        assert!(
            Arc::ptr_eq(key1, key2),
            "interner must share the same Arc<str> across edges; got distinct allocations"
        );

        // Same check on a key inserted via the second slot.
        let kind1 = edge1
            .properties
            .keys()
            .find(|k| &***k == "event_kind")
            .expect("event_kind key on first edge");
        let kind2 = edge2
            .properties
            .keys()
            .find(|k| &***k == "event_kind")
            .expect("event_kind key on second edge");
        assert!(
            Arc::ptr_eq(kind1, kind2),
            "interner must share event_kind Arc<str> across edges"
        );
    }
}
