graph_explorer_core/

preview.rs

//! The provisional layer: one node's discovered neighborhood, shown to the user
//! but not yet adopted into the committed `Graph`.
//!
//! Discovery is asynchronous, so a preview can be superseded before its data
//! arrives. Every preview therefore carries a monotonic `generation`, and
//! `fill` refuses anything but the live one — a late arrival can never
//! repopulate a layer the user has already moved on from. This guard is
//! independent of (and holds even without) the `AbortSignal` the host fetcher
//! receives.
//!
//! Pure: no I/O, no wasm, no geometry. Positions for provisional nodes are the
//! caller's business.
//!
//! The generation is deliberately write-only from the outside: there is no
//! accessor for it. `begin` hands the caller a token; `fill` only accepts
//! that exact token back. A getter would let a caller "refresh" the token by
//! reading it right before calling `fill`, which launders away the guard
//! this type exists to provide.

use crate::{Aggregate, Edge, Graph, NeighborResult, Node, NodeId};

#[derive(Debug, Default)]
pub struct PreviewLayer {
    generation: u64,
    anchor: Option<NodeId>,
    nodes: Vec<Node>,
    edges: Vec<Edge>,
    aggregates: Vec<Aggregate>,
}

impl PreviewLayer {
    pub fn new() -> Self { Self::default() }

    /// Supersede any live preview and open a new one anchored at `anchor`.
    /// Returns the new generation, which the caller must present to `fill`.
    pub fn begin(&mut self, anchor: NodeId) -> u64 {
        self.generation += 1;
        self.anchor = Some(anchor);
        self.nodes.clear();
        self.edges.clear();
        self.aggregates.clear();
        self.generation
    }

    pub fn anchor(&self) -> Option<&NodeId> { self.anchor.as_ref() }
    pub fn nodes(&self) -> &[Node] { &self.nodes }
    pub fn edges(&self) -> &[Edge] { &self.edges }
    pub fn aggregates(&self) -> &[Aggregate] { &self.aggregates }

    /// How many nodes are provisionally shown. Edge-only and aggregate-only
    /// previews report 0 here but are not `is_empty`. Deliberately not named
    /// `len`: pairing a node-only count with an `is_empty` that also considers
    /// edges and aggregates would violate the usual
    /// `is_empty() == (len() == 0)` assumption.
    pub fn node_count(&self) -> usize { self.nodes.len() }

    /// Whether the layer holds nothing at all. Note this considers aggregates
    /// as well, so an aggregate-only neighborhood is NOT empty even though
    /// `node_count()` is 0 — there is something for a host to act on.
    pub fn is_empty(&self) -> bool {
        self.nodes.is_empty() && self.edges.is_empty() && self.aggregates.is_empty()
    }

    /// Whether `take_all` would actually hand anything back. Distinct from
    /// `is_empty()`, which also counts aggregates: an aggregate is reported to
    /// a host, never committed into the `Graph`, so a commit path must ask
    /// this instead.
    pub fn has_committable(&self) -> bool {
        !self.nodes.is_empty() || !self.edges.is_empty()
    }

    /// Whether `id` is among the provisional nodes. Consults nodes only —
    /// an id that appears solely as an edge endpoint is not "contained".
    pub fn contains(&self, id: &str) -> bool { self.nodes.iter().any(|n| n.id == id) }

    /// Adopt a neighbor result, keeping only what `committed` lacks. Returns
    /// false and changes nothing when `gen` is not the live generation, or
    /// when the layer was never `begin`-ed (a fresh layer's generation is 0,
    /// which must not be fillable by passing 0).
    ///
    /// Replaces rather than appends: a preview is one page of one neighborhood.
    pub fn fill(&mut self, gen: u64, res: &NeighborResult, committed: &Graph) -> bool {
        let Some(anchor) = self.anchor.clone() else { return false };
        if gen != self.generation {
            return false;
        }
        // Keep the first occurrence of each id: a provider returning a node
        // twice must not over-count the ghost/placement fan shown to the user.
        let mut seen = std::collections::HashSet::new();
        let nodes: Vec<Node> = res
            .nodes
            .iter()
            .filter(|n| committed.node(&n.id).is_none())
            .filter(|n| seen.insert(n.id.clone()))
            .cloned()
            .collect();
        // The node filter must run first: an edge is only safe to admit once
        // we know which nodes are actually going to be in this preview.
        let retained: std::collections::HashSet<&str> = nodes.iter().map(|n| n.id.as_str()).collect();
        let known = |id: &str| committed.node(id).is_some() || retained.contains(id);
        self.edges = res
            .edges
            .iter()
            // An edge whose endpoints are both committed but whose id is not is
            // still new information, so membership is tested by edge id.
            .filter(|e| committed.edge(&e.id).is_none())
            // A provider can page nodes and return edges spanning into the next
            // page; an edge with an endpoint that is neither committed nor
            // among the nodes just kept would dangle once `take_all` commits
            // it, which `Graph::from_json` explicitly refuses to accept.
            .filter(|e| known(&e.source) && known(&e.target))
            .cloned()
            .collect();
        self.nodes = nodes;
        // Aggregates are kept whole — there is nothing to filter against, since
        // a group is not a node. `parent` is stamped from the anchor rather
        // than believed from the wire.
        self.aggregates = res.aggregates.iter().cloned().map(|mut a| {
            a.parent = anchor.clone();
            a
        }).collect();
        true
    }

    /// Move one provisional node out of the layer, along with the incident
    /// edges that can safely follow it — those whose *other* endpoint is
    /// already committed (or the node itself, for a self-loop). An edge to
    /// another provisional node stays behind rather than dangling.
    pub fn promote(&mut self, id: &str, committed: &Graph) -> Option<(Node, Vec<Edge>)> {
        let idx = self.nodes.iter().position(|n| n.id == id)?;
        let node = self.nodes.remove(idx);
        let mut taken = Vec::new();
        self.edges.retain(|e| {
            if e.source != id && e.target != id {
                return true;
            }
            let other = if e.source == id { &e.target } else { &e.source };
            if other == id || committed.node(other).is_some() {
                taken.push(e.clone());
                false
            } else {
                true
            }
        });
        Some((node, taken))
    }

    /// Drain the committable content (nodes and edges), for an explicit
    /// commit. Retires the generation exactly as `clear` does — a fetch
    /// still in flight for this preview must not repopulate it after the
    /// fact. The **aggregates and the anchor are retained**: they are
    /// reported-not-committed offers, and accepting the ghosts is no reason
    /// to withdraw the "expand this group" affordance the host may be
    /// showing. They die with the preview itself (next `begin`, supersede,
    /// or `clear`), not with the commit.
    pub fn take_all(&mut self) -> (Vec<Node>, Vec<Edge>) {
        self.generation += 1;
        (std::mem::take(&mut self.nodes), std::mem::take(&mut self.edges))
    }

    /// Discard without committing. Retires the generation as well: a discard is
    /// exactly the case where an in-flight fetch is about to land, and refilling
    /// the layer the user just dismissed would be the worst possible outcome.
    pub fn clear(&mut self) {
        self.generation += 1;
        self.anchor = None;
        self.nodes.clear();
        self.edges.clear();
        self.aggregates.clear();
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::{Aggregate, Edge, Graph, GroupBy, NeighborResult, Node, QueryParams};

    fn committed(node_ids: &[&str], edges: &[(&str, &str, &str)]) -> Graph {
        let mut g = Graph::default();
        for id in node_ids { g.add_node(Node::new(*id)); }
        for (id, s, t) in edges { g.add_edge(Edge::new(*id, *s, *t)); }
        g
    }

    fn result(nodes: &[&str], edges: &[(&str, &str, &str)]) -> NeighborResult {
        NeighborResult {
            nodes: nodes.iter().map(|id| Node::new(*id)).collect(),
            edges: edges.iter().map(|(id, s, t)| Edge::new(*id, *s, *t)).collect(),
            aggregates: vec![],
            next: None,
            pending: false,
        }
    }

    fn result_with_agg(nodes: &[&str], aggs: &[(&str, &str, u64)]) -> NeighborResult {
        NeighborResult {
            nodes: nodes.iter().map(|id| Node::new(*id)).collect(),
            edges: vec![],
            aggregates: aggs.iter().map(|(id, value, count)| Aggregate {
                id: (*id).into(),
                parent: String::new(),
                group_by: GroupBy::Label,
                value: (*value).into(),
                relationships: vec![],
                count: *count,
                query: QueryParams { limit: 8, cursor: None },
            }).collect(),
            next: None,
            pending: false,
        }
    }

    #[test]
    fn begin_bumps_the_generation_and_clears_prior_contents() {
        let g = committed(&["a"], &[]);
        let mut p = PreviewLayer::new();
        let g1 = p.begin("a".into());
        assert!(p.fill(g1, &result(&["b"], &[]), &g));
        assert_eq!(p.node_count(), 1);

        let g2 = p.begin("b".into());
        assert!(g2 > g1, "each preview gets a fresh generation");
        assert!(p.is_empty(), "beginning a new preview clears the old contents");
        assert_eq!(p.anchor(), Some(&"b".to_string()));
    }

    #[test]
    fn fill_with_a_stale_generation_is_rejected_and_changes_nothing() {
        let g = committed(&["a"], &[]);
        let mut p = PreviewLayer::new();
        let stale = p.begin("a".into());
        p.begin("b".into());

        assert!(!p.fill(stale, &result(&["x", "y"], &[]), &g), "stale fill must be refused");
        assert!(p.is_empty(), "a refused fill must not leak nodes into the layer");
    }

    #[test]
    fn fill_drops_nodes_already_committed() {
        let g = committed(&["a", "known"], &[]);
        let mut p = PreviewLayer::new();
        let gen = p.begin("a".into());
        p.fill(gen, &result(&["known", "fresh"], &[]), &g);

        let ids: Vec<&str> = p.nodes().iter().map(|n| n.id.as_str()).collect();
        assert_eq!(ids, vec!["fresh"]);
    }

    #[test]
    fn fill_keeps_a_new_edge_between_two_committed_nodes() {
        // Both endpoints are known but the relationship is not: that is still
        // new information and belongs in the preview.
        let g = committed(&["a", "b"], &[]);
        let mut p = PreviewLayer::new();
        let gen = p.begin("a".into());
        p.fill(gen, &result(&[], &[("e1", "a", "b")]), &g);

        assert_eq!(p.edges().len(), 1);
        assert!(!p.is_empty(), "edge-only content still counts as a preview");
    }

    #[test]
    fn fill_drops_edges_already_committed() {
        let g = committed(&["a", "b"], &[("e1", "a", "b")]);
        let mut p = PreviewLayer::new();
        let gen = p.begin("a".into());
        p.fill(gen, &result(&[], &[("e1", "a", "b")]), &g);
        assert!(p.edges().is_empty());
    }

    #[test]
    fn promote_takes_the_node_and_only_safely_committable_edges() {
        // "a" is committed; "b" and "c" are provisional siblings.
        // Promoting "b" may take b–a (other endpoint committed) but NOT b–c,
        // which would dangle to a node that does not exist yet.
        let g = committed(&["a"], &[]);
        let mut p = PreviewLayer::new();
        let gen = p.begin("a".into());
        p.fill(gen, &result(&["b", "c"], &[("ab", "a", "b"), ("bc", "b", "c")]), &g);

        let (node, edges) = p.promote("b", &g).expect("b is provisional");
        assert_eq!(node.id, "b");
        let ids: Vec<&str> = edges.iter().map(|e| e.id.as_str()).collect();
        assert_eq!(ids, vec!["ab"]);

        assert!(!p.contains("b"), "a promoted node leaves the preview");
        assert!(p.contains("c"), "its sibling stays behind");
        assert_eq!(p.edges().len(), 1, "b–c stays until c is committed too");
    }

    #[test]
    fn fill_drops_an_edge_into_an_unknown_node() {
        // The provider paged nodes but returned an edge spanning into the next
        // page: neither endpoint exists yet, so admitting it would hand the
        // caller a dangling edge once take_all commits it.
        let g = committed(&["a"], &[]);
        let mut p = PreviewLayer::new();
        let gen = p.begin("a".into());
        p.fill(gen, &result(&["b"], &[("bz", "b", "z")]), &g);
        assert!(p.edges().is_empty(), "an edge to an unknown node must not survive fill");
    }

    #[test]
    fn fill_on_a_fresh_layer_is_rejected() {
        // generation starts at 0 with no anchor; fill(0, ..) must not be able
        // to sneak data into a layer that was never begun.
        let g = committed(&["a"], &[]);
        let mut p = PreviewLayer::new();
        assert!(!p.fill(0, &result(&["b"], &[]), &g));
        assert!(p.is_empty());
    }

    #[test]
    fn fill_dedupes_duplicate_node_ids() {
        let g = committed(&["a"], &[]);
        let mut p = PreviewLayer::new();
        let gen = p.begin("a".into());
        p.fill(gen, &result(&["b", "b"], &[]), &g);
        assert_eq!(p.node_count(), 1, "duplicate ids collapse to one entry");
    }

    #[test]
    fn promote_of_an_unknown_or_already_promoted_id_is_none() {
        let g = committed(&["a"], &[]);
        let mut p = PreviewLayer::new();
        let gen = p.begin("a".into());
        p.fill(gen, &result(&["b"], &[]), &g);

        assert!(p.promote("nope", &g).is_none());
        assert!(p.promote("b", &g).is_some());
        assert!(p.promote("b", &g).is_none(), "promoting twice is not possible");
    }

    #[test]
    fn take_all_drains_the_layer() {
        let g = committed(&["a"], &[]);
        let mut p = PreviewLayer::new();
        let gen = p.begin("a".into());
        p.fill(gen, &result(&["b", "c"], &[("ab", "a", "b")]), &g);

        let (nodes, edges) = p.take_all();
        assert_eq!(nodes.len(), 2);
        assert_eq!(edges.len(), 1);
        assert!(p.is_empty(), "take_all leaves nothing behind");
        assert!(p.take_all().0.is_empty(), "draining twice yields nothing");
    }

    #[test]
    fn take_all_retains_aggregates_and_anchor_while_clearing_committables() {
        // A commit accepts the ghosts but the aggregate offers (and the
        // anchor that makes them actionable) must survive it: the host's
        // "expand this group" affordance is not tied to the ghosts' fate.
        let g = committed(&["a"], &[]);
        let mut p = PreviewLayer::new();
        let gen = p.begin("a".into());
        p.fill(gen, &result_with_agg(&["b"], &[("agg:X", "X", 3)]), &g);

        let (nodes, edges) = p.take_all();
        assert_eq!(nodes.len(), 1, "the committable node was drained");
        assert!(edges.is_empty());

        assert!(p.nodes().is_empty(), "committable nodes are gone from the layer");
        assert!(p.edges().is_empty(), "committable edges are gone from the layer");
        assert_eq!(p.aggregates().len(), 1, "the aggregate offer survives the commit");
        assert_eq!(p.anchor(), Some(&"a".to_string()), "the anchor survives too");
        assert!(!p.has_committable(), "nothing left to commit");
        assert!(!p.is_empty(), "the surviving aggregate still counts as something to report");
    }

    #[test]
    fn take_all_still_retires_the_generation() {
        // Retaining the aggregates must not reopen the door for a late fetch:
        // cancellation safety is untouched by this change.
        let g = committed(&["a"], &[]);
        let mut p = PreviewLayer::new();
        let gen = p.begin("a".into());
        p.fill(gen, &result_with_agg(&["b"], &[("agg:X", "X", 3)]), &g);
        p.take_all();

        assert!(!p.fill(gen, &result(&["c"], &[]), &g), "a late arrival must still be blocked");
    }

    #[test]
    fn clear_after_take_all_drops_the_retained_aggregates() {
        // The aggregates retained by take_all still die with the preview
        // itself — the next begin, a supersede, or an explicit clear.
        let g = committed(&["a"], &[]);
        let mut p = PreviewLayer::new();
        let gen = p.begin("a".into());
        p.fill(gen, &result_with_agg(&["b"], &[("agg:X", "X", 3)]), &g);
        p.take_all();
        p.clear();

        assert!(p.aggregates().is_empty());
        assert_eq!(p.anchor(), None);
    }

    #[test]
    fn fill_after_take_all_is_rejected() {
        // take_all is a commit: a fetch still in flight for the drained preview
        // must not repopulate it afterwards.
        let g = committed(&["a"], &[]);
        let mut p = PreviewLayer::new();
        let gen = p.begin("a".into());
        p.fill(gen, &result(&["b"], &[]), &g);
        p.take_all();

        assert!(!p.fill(gen, &result(&["c"], &[]), &g));
        assert!(p.is_empty(), "no aggregates were ever filled, so nothing is retained here");
        assert_eq!(
            p.anchor(),
            Some(&"a".to_string()),
            "the anchor survives a commit — it only dies with the next begin/supersede/clear"
        );
    }

    #[test]
    fn fill_after_promote_is_still_accepted() {
        // Promotion shrinks the preview but does not end it: the same fetch
        // page can still land on what remains.
        let g = committed(&["a"], &[]);
        let mut p = PreviewLayer::new();
        let gen = p.begin("a".into());
        p.fill(gen, &result(&["b"], &[]), &g);
        p.promote("b", &g);

        assert!(p.fill(gen, &result(&["c"], &[]), &g));
        assert!(p.contains("c"));
    }

    #[test]
    fn promote_takes_a_self_loop_on_the_promoted_node() {
        let g = committed(&["a"], &[]);
        let mut p = PreviewLayer::new();
        let gen = p.begin("a".into());
        p.fill(gen, &result(&["b"], &[("loop", "b", "b")]), &g);

        let (_, edges) = p.promote("b", &g).expect("b is provisional");
        let ids: Vec<&str> = edges.iter().map(|e| e.id.as_str()).collect();
        assert_eq!(ids, vec!["loop"], "a self-loop travels with its node");
    }

    #[test]
    fn promote_leaves_non_incident_edges_untouched() {
        // "c" and "d" are both provisional and unrelated to "b".
        let g = committed(&["a"], &[]);
        let mut p = PreviewLayer::new();
        let gen = p.begin("a".into());
        p.fill(gen, &result(&["b", "c", "d"], &[("cd", "c", "d")]), &g);

        let (_, taken) = p.promote("b", &g).expect("b is provisional");
        assert!(taken.is_empty(), "b has no incident edges");
        assert_eq!(p.edges().len(), 1, "c-d is untouched by promoting b");
    }

    #[test]
    fn clear_retires_the_generation_so_a_late_arrival_cannot_refill_it() {
        // Without the bump, a discarded preview would be silently repopulated
        // by the fetch it was discarded to escape.
        let g = committed(&["a"], &[]);
        let mut p = PreviewLayer::new();
        let gen = p.begin("a".into());
        p.clear();

        assert!(!p.fill(gen, &result(&["b"], &[]), &g));
        assert!(p.is_empty());
        assert_eq!(p.anchor(), None);
    }

    #[test]
    fn contains_reports_membership_of_the_provisional_set() {
        let g = committed(&["a"], &[]);
        let mut p = PreviewLayer::new();
        let gen = p.begin("a".into());
        p.fill(gen, &result(&["b"], &[]), &g);
        assert!(p.contains("b"));
        assert!(!p.contains("a"), "the anchor is committed, not provisional");
    }

    #[test]
    fn fill_retains_aggregates_and_stamps_their_parent() {
        // The parent is what makes an aggregate actionable; it is derived from
        // the anchor we queried, never read from the wire.
        let g = committed(&["a"], &[]);
        let mut p = PreviewLayer::new();
        let gen = p.begin("a".into());
        p.fill(gen, &result_with_agg(&[], &[("agg:Sorcery", "Sorcery", 32)]), &g);

        assert_eq!(p.aggregates().len(), 1);
        assert_eq!(p.aggregates()[0].parent, "a", "stamped from the anchor");
        assert_eq!(p.aggregates()[0].count, 32);
    }

    #[test]
    fn fill_overwrites_a_wire_supplied_parent_with_the_anchor() {
        // A provider that dutifully fills in `parent` itself must still be
        // overruled: stamping is unconditional, not a fallback for an omitted
        // value. Otherwise "derived, never trusted" only holds by accident.
        let g = committed(&["a"], &[]);
        let mut p = PreviewLayer::new();
        let gen = p.begin("a".into());
        let res = NeighborResult {
            nodes: vec![],
            edges: vec![],
            aggregates: vec![Aggregate {
                id: "agg:Sorcery".into(),
                parent: "somewhere-else".into(),
                group_by: GroupBy::Label,
                value: "Sorcery".into(),
                relationships: vec![],
                count: 32,
                query: QueryParams { limit: 8, cursor: None },
            }],
            next: None,
            pending: false,
        };
        p.fill(gen, &res, &g);

        assert_eq!(p.aggregates()[0].parent, "a", "the anchor overwrites whatever the wire sent");
    }

    #[test]
    fn fill_keeps_aggregates_whole_even_if_their_id_collides_with_a_committed_node() {
        // An aggregate is not a node, so the node-existence filter must never
        // reach it: a group is fetchable, actionable content in its own right,
        // regardless of what ids happen to already be committed.
        let g = committed(&["a", "agg:Sorcery"], &[]);
        let mut p = PreviewLayer::new();
        let gen = p.begin("a".into());
        p.fill(gen, &result_with_agg(&[], &[("agg:Sorcery", "Sorcery", 32)]), &g);

        assert_eq!(p.aggregates().len(), 1, "kept whole, not filtered like a node");
    }

    #[test]
    fn an_aggregate_only_neighborhood_is_not_empty() {
        // This is the reported bug: walking to a fully-aggregated node showed
        // zero ghosts and reported nothing, so the graph looked broken.
        let g = committed(&["a"], &[]);
        let mut p = PreviewLayer::new();
        let gen = p.begin("a".into());
        p.fill(gen, &result_with_agg(&[], &[("agg:Sorcery", "Sorcery", 32)]), &g);

        assert_eq!(p.node_count(), 0, "no individual nodes");
        assert!(!p.is_empty(), "but the layer is NOT empty — there is something to report");
    }

    #[test]
    fn begin_and_clear_drop_aggregates_but_take_all_does_not() {
        let g = committed(&["a"], &[]);
        let mut p = PreviewLayer::new();
        let gen = p.begin("a".into());
        p.fill(gen, &result_with_agg(&["b"], &[("agg:X", "X", 3)]), &g);
        assert_eq!(p.aggregates().len(), 1);

        p.begin("b".into());
        assert!(p.aggregates().is_empty(), "a new preview starts clean");

        let gen2 = p.begin("a".into());
        p.fill(gen2, &result_with_agg(&["c"], &[("agg:X", "X", 3)]), &g);
        p.take_all();
        assert_eq!(
            p.aggregates().len(),
            1,
            "committing the ghosts retains the aggregate offers — they die with the preview, not the commit"
        );

        let gen3 = p.begin("a".into());
        p.fill(gen3, &result_with_agg(&["d"], &[("agg:X", "X", 3)]), &g);
        p.clear();
        assert!(p.aggregates().is_empty(), "discarding drops them");
    }

    #[test]
    fn has_committable_distinguishes_reportable_aggregates_from_committable_content() {
        let g = committed(&["a"], &[]);

        // Aggregate-only: something to report, nothing to commit.
        let mut p = PreviewLayer::new();
        let gen = p.begin("a".into());
        p.fill(gen, &result_with_agg(&[], &[("agg:Sorcery", "Sorcery", 32)]), &g);
        assert!(!p.is_empty(), "an aggregate is reportable");
        assert!(!p.has_committable(), "but an aggregate is never committed into the Graph");

        // Node-only: both reportable and committable.
        let mut p = PreviewLayer::new();
        let gen = p.begin("a".into());
        p.fill(gen, &result(&["b"], &[]), &g);
        assert!(!p.is_empty());
        assert!(p.has_committable(), "a provisional node is committable");

        // Edge-only: both reportable and committable.
        let g2 = committed(&["a", "b"], &[]);
        let mut p = PreviewLayer::new();
        let gen = p.begin("a".into());
        p.fill(gen, &result(&[], &[("e1", "a", "b")]), &g2);
        assert!(!p.is_empty());
        assert!(p.has_committable(), "a provisional edge is committable");
    }

    #[test]
    fn a_stale_fill_does_not_leak_aggregates() {
        let g = committed(&["a"], &[]);
        let mut p = PreviewLayer::new();
        let stale = p.begin("a".into());
        p.begin("b".into());
        assert!(!p.fill(stale, &result_with_agg(&[], &[("agg:X", "X", 9)]), &g));
        assert!(p.aggregates().is_empty());
    }
}