Skip to main content

core_api/
algo.rs

1//! Graph algorithms: PageRank, weakly-connected components, degree centrality.
2//!
3//! ## Dependency rule note
4//!
5//! This module lives in `core-api` (not `core-query`) because it must read the
6//! *unified topology* — manual edges plus derived edges written by the rule
7//! engine via `GraphMut`. `core-query` has no dependency on `core-rules` and
8//! therefore cannot see derived provenance.  `GraphDb` fields are private; the
9//! algorithms are pure functions called from `GraphDb` methods that pass in a
10//! [`TopologyView`] over the in-memory overlay **and** the mmap V8 base. Reading
11//! the view (not the bare overlay `Topology`) is required: after a snapshot
12//! reopen the derived edges live in the base, and using the overlay alone would
13//! report zero degree/rank for every node.
14//!
15//! ## When to use views vs `degree_centrality`
16//!
17//! `degree_centrality` is a **one-shot compute**: call it, get a snapshot sorted
18//! by degree, done.  It does not persist anywhere and is not maintained as
19//! properties change.  Use it for offline analysis, ranking a batch, or feeding
20//! `write_scores` once.
21//!
22//! A **Degree materialized view** (`ViewDef { kind: AggFn::Degree, … }`) is
23//! *maintained incrementally*: every `insert_edge` / `delete_edge` /
24//! `insert_node` re-computes just the affected node's count and stores it as a
25//! live property.  Use it when you need the degree of individual nodes at query
26//! time with zero latency (e.g. `MATCH (n) WHERE n.out_degree > 5 RETURN n`).
27//!
28//! Rule of thumb: if you need the top-K by degree once → `degree_centrality`;
29//! if you need the degree of every node available in every Cypher query →
30//! create a Degree view.
31
32use core_query::Dir;
33use core_storage::v8::seam::TopologyView;
34use core_storage::{Direction, EdgePropsView, IdMap, Interner, Value};
35use serde::{Deserialize, Serialize};
36use std::collections::BTreeMap;
37use std::time::{Duration, Instant};
38
39// ---------------------------------------------------------------------------
40// Shared helpers
41// ---------------------------------------------------------------------------
42
43/// Collect the dense list of live node ids and their string keys.
44///
45/// Returns `(ids, keys)` where `ids[i]` is the internal u32 id for `keys[i]`.
46/// Tombstoned slots are skipped.  Stable order: ascending internal id.
47fn live_nodes(idmap: &IdMap, labels: &[u32]) -> (Vec<u32>, Vec<String>) {
48    let n = idmap.len() as u32;
49    let mut ids = Vec::new();
50    let mut keys = Vec::new();
51    for id in 0..n {
52        let Some(key) = idmap.key_of(id) else {
53            continue;
54        };
55        let Some(&sym) = labels.get(id as usize) else {
56            continue;
57        };
58        if sym == u32::MAX {
59            continue; // tombstoned
60        }
61        ids.push(id);
62        keys.push(key.to_string());
63    }
64    (ids, keys)
65}
66
67/// Resolve an optional edge-type name to its interned symbol.
68///
69/// Returns `None` if `edge_type` is `Some(name)` that is not interned (meaning
70/// no edges of that type exist).  Returns `Some(None)` when `edge_type` is
71/// `None` (all types).
72fn resolve_etype(syms: &Interner, edge_type: Option<&str>) -> Option<Option<u32>> {
73    match edge_type {
74        None => Some(None), // all etypes
75        Some(name) => {
76            let sym = syms.get(name)?; // not interned → no such edges
77            Some(Some(sym))
78        }
79    }
80}
81
82/// Iterate over etypes in the topology, optionally filtered to a single etype.
83fn etypes_filtered(topo: &TopologyView, filter: Option<u32>) -> Vec<u32> {
84    match filter {
85        Some(sym) => {
86            // Only include if the etype actually exists.
87            let all: Vec<u32> = topo.etypes().collect();
88            if all.contains(&sym) {
89                vec![sym]
90            } else {
91                vec![]
92            }
93        }
94        None => topo.etypes().collect(),
95    }
96}
97
98/// Resolve a list of edge-type names to their interned symbols, in the given
99/// order, de-duplicated. Unresolved names (not interned — no such edges
100/// exist) are silently skipped. An empty `names` means "all edge types".
101fn resolve_etypes_multi(syms: &Interner, topo: &TopologyView, names: &[String]) -> Vec<u32> {
102    if names.is_empty() {
103        return topo.etypes().collect();
104    }
105    let mut out = Vec::new();
106    for name in names {
107        if let Some(sym) = syms.get(name) {
108            if !out.contains(&sym) {
109                out.push(sym);
110            }
111        }
112    }
113    out
114}
115
116/// Like [`live_nodes`] but optionally restricted to nodes carrying `label`.
117/// `None` includes every live node; `Some(name)` not interned yields an empty
118/// result (no such nodes exist).
119fn live_nodes_for_label(
120    idmap: &IdMap,
121    syms: &Interner,
122    labels: &[u32],
123    label: Option<&str>,
124) -> (Vec<u32>, Vec<String>) {
125    let want = match label {
126        None => None,
127        Some(name) => match syms.get(name) {
128            Some(sym) => Some(sym),
129            None => return (Vec::new(), Vec::new()),
130        },
131    };
132    let n = idmap.len() as u32;
133    let mut ids = Vec::new();
134    let mut keys = Vec::new();
135    for id in 0..n {
136        let Some(key) = idmap.key_of(id) else {
137            continue;
138        };
139        let Some(&sym) = labels.get(id as usize) else {
140            continue;
141        };
142        if sym == u32::MAX {
143            continue; // tombstoned
144        }
145        if let Some(want_sym) = want {
146            if sym != want_sym {
147                continue;
148            }
149        }
150        ids.push(id);
151        keys.push(key.to_string());
152    }
153    (ids, keys)
154}
155
156/// Resolve the weight of edge `(etype, src, dst)`.
157///
158/// `weight_prop`: read this numeric edge property; missing or non-numeric
159/// values fall back to `1.0`. `None` treats every edge as weight `1.0`.
160/// `min_weight`: drop the edge (return `None`) when its resolved weight is
161/// below this threshold — applied regardless of whether `weight_prop` is set.
162fn edge_weight(
163    edge_props: &EdgePropsView,
164    etype: u32,
165    src: u32,
166    dst: u32,
167    weight_prop: Option<&str>,
168    min_weight: Option<f64>,
169) -> Option<f64> {
170    let w = match weight_prop {
171        None => 1.0,
172        Some(prop) => match edge_props.get(etype, src, dst, prop) {
173            Some(Value::Float(f)) => f,
174            Some(Value::Int(i)) => i as f64,
175            _ => 1.0,
176        },
177    };
178    match min_weight {
179        Some(min) if w < min => None,
180        _ => Some(w),
181    }
182}
183
184// ---------------------------------------------------------------------------
185// PageRank
186// ---------------------------------------------------------------------------
187
188/// Configuration for [`GraphDb::pagerank`].
189#[derive(Debug, Clone, Serialize, Deserialize)]
190#[serde(default)]
191pub struct PageRankConfig {
192    /// Damping factor (probability of following an edge, not teleporting).
193    /// Default 0.85.
194    pub damping: f64,
195    /// Maximum number of power-iteration steps. Default 50.
196    pub max_iters: u32,
197    /// Convergence tolerance (L1 norm over all nodes). Default 1e-6.
198    pub tol: f64,
199    /// Restrict edges to this type. `None` uses all edge types (unified topology).
200    pub edge_type: Option<String>,
201    /// Edge direction to follow. `Dir::Out` follows out-edges (standard web
202    /// PageRank); `Dir::In` follows in-edges (authority scores); `Dir::Both`
203    /// treats all edges as undirected.
204    pub direction: AlgoDir,
205    /// Wall-clock budget (milliseconds) for the HTTP server endpoint.
206    /// `0` means no budget (run to convergence or `max_iters`).
207    pub budget_ms: u64,
208    /// Read this edge property as the edge weight; missing or non-numeric
209    /// values fall back to `1.0`. `None` treats every edge as weight `1.0`
210    /// (mass distributes uniformly across out-edges, as before).
211    pub weight_prop: Option<String>,
212    /// Drop edges whose resolved weight is below this threshold before the
213    /// algorithm runs. Applied whether or not `weight_prop` is set.
214    pub min_weight: Option<f64>,
215}
216
217impl Default for PageRankConfig {
218    fn default() -> Self {
219        Self {
220            damping: 0.85,
221            max_iters: 50,
222            tol: 1e-6,
223            edge_type: None,
224            direction: AlgoDir::Out,
225            budget_ms: 5_000,
226            weight_prop: None,
227            min_weight: None,
228        }
229    }
230}
231
232/// Result of [`GraphDb::pagerank`].
233#[derive(Debug, Clone, Serialize, Deserialize)]
234pub struct PageRankReport {
235    /// Node keys and their PageRank scores.  Sorted: score descending, key
236    /// ascending on ties (deterministic).
237    pub scores: Vec<(String, f64)>,
238    /// `true` if the algorithm converged before `max_iters` and before any time
239    /// budget fired.  `false` means scores are still valid but partial — more
240    /// iterations would refine them.
241    pub converged: bool,
242}
243
244/// Direction semantics for algo methods (mirrors `Dir` but serializable).
245#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
246#[serde(rename_all = "lowercase")]
247pub enum AlgoDir {
248    /// Follow outgoing edges only (standard directed PageRank / out-degree).
249    Out,
250    /// Follow incoming edges only (in-degree / authority score).
251    In,
252    /// Treat edges as undirected: Out ∪ In.
253    Both,
254}
255
256impl From<Dir> for AlgoDir {
257    fn from(d: Dir) -> Self {
258        match d {
259            Dir::Out => AlgoDir::Out,
260            Dir::In => AlgoDir::In,
261            Dir::Both => AlgoDir::Both,
262        }
263    }
264}
265
266/// Run PageRank on the unified topology.
267///
268/// Returns a [`PageRankReport`] with scores sorted descending (ties: key asc).
269pub(crate) fn pagerank(
270    topo: &TopologyView,
271    idmap: &IdMap,
272    syms: &Interner,
273    labels: &[u32],
274    edge_props: &EdgePropsView,
275    config: &PageRankConfig,
276) -> PageRankReport {
277    let weighted = config.weight_prop.is_some() || config.min_weight.is_some();
278    let deadline = if config.budget_ms > 0 {
279        Some(Instant::now() + Duration::from_millis(config.budget_ms))
280    } else {
281        None
282    };
283
284    let (node_ids, node_keys) = live_nodes(idmap, labels);
285    let n = node_ids.len();
286
287    if n == 0 {
288        return PageRankReport {
289            scores: Vec::new(),
290            converged: true,
291        };
292    }
293
294    // Map internal id → compact index for fast array access.
295    let max_id = topo.etypes().count(); // just an upper bound check hint
296    let _ = max_id;
297    let mut id_to_idx: BTreeMap<u32, usize> = BTreeMap::new();
298    for (i, &id) in node_ids.iter().enumerate() {
299        id_to_idx.insert(id, i);
300    }
301
302    // Resolve etype filter.
303    let etype_filter = match resolve_etype(syms, config.edge_type.as_deref()) {
304        None => {
305            // Edge type specified but not in graph → no edges, PR is uniform.
306            let score = 1.0 / n as f64;
307            let mut scores: Vec<(String, f64)> =
308                node_keys.iter().map(|k| (k.clone(), score)).collect();
309            scores.sort_by(|(ka, sa), (kb, sb)| {
310                sb.partial_cmp(sa)
311                    .unwrap_or(std::cmp::Ordering::Equal)
312                    .then(ka.cmp(kb))
313            });
314            return PageRankReport {
315                scores,
316                converged: true,
317            };
318        }
319        Some(f) => f,
320    };
321
322    let etypes = etypes_filtered(topo, etype_filter);
323
324    // Build adjacency list (compact index): for each compact node, which
325    // compact nodes does it "send" rank to (based on direction), and with
326    // what weight.  Unweighted mode dedups parallel edges to the same
327    // neighbor (matches pre-weight behavior exactly); weighted mode sums the
328    // resolved weight of every qualifying edge instance (accumulating across
329    // parallel edges / multiple edge types).
330    let mut send_to: Vec<Vec<(usize, f64)>> = vec![Vec::new(); n];
331
332    for &et in &etypes {
333        for (i, &id) in node_ids.iter().enumerate() {
334            let dirs: &[Direction] = match config.direction {
335                AlgoDir::Out => &[Direction::Out],
336                AlgoDir::In => &[Direction::In],
337                AlgoDir::Both => &[Direction::Out, Direction::In],
338            };
339            for &dir in dirs {
340                for &nbr in topo.neighbors(et, dir, id).as_ref() {
341                    let Some(&j) = id_to_idx.get(&nbr) else {
342                        continue;
343                    };
344                    if weighted {
345                        let Some(w) = edge_weight(
346                            edge_props,
347                            et,
348                            id,
349                            nbr,
350                            config.weight_prop.as_deref(),
351                            config.min_weight,
352                        ) else {
353                            continue; // filtered by min_weight
354                        };
355                        if let Some(entry) = send_to[i].iter_mut().find(|(k, _)| *k == j) {
356                            entry.1 += w;
357                        } else {
358                            send_to[i].push((j, w));
359                        }
360                    } else if !send_to[i].iter().any(|(k, _)| *k == j) {
361                        send_to[i].push((j, 1.0));
362                    }
363                }
364            }
365        }
366    }
367
368    // Build receive_from[j] = list of (i, share) that send to j, where share
369    // is i's outgoing weight to j normalized by i's total outgoing weight.
370    // Also track dangling nodes (no outgoing weight).
371    let mut receive_from: Vec<Vec<(usize, f64)>> = vec![Vec::new(); n];
372    let mut dangling: Vec<usize> = Vec::new();
373
374    for (i, send) in send_to.iter().enumerate() {
375        let out_weight: f64 = send.iter().map(|(_, w)| w).sum();
376        if send.is_empty() || out_weight <= 0.0 {
377            dangling.push(i);
378        } else {
379            for &(j, w) in send {
380                receive_from[j].push((i, w / out_weight));
381            }
382        }
383    }
384
385    // Power iteration.
386    let nf = n as f64;
387    let d = config.damping;
388    let teleport = (1.0 - d) / nf;
389    let mut pr: Vec<f64> = vec![1.0 / nf; n];
390    let mut converged = false;
391
392    for _iter in 0..config.max_iters {
393        // Check time budget between iterations.
394        if let Some(dl) = deadline {
395            if Instant::now() >= dl {
396                break;
397            }
398        }
399
400        // Sum PR leaked by dangling nodes → distribute uniformly.
401        let dangling_sum: f64 = dangling.iter().map(|&i| pr[i]).sum::<f64>() * d / nf;
402
403        let mut new_pr = vec![teleport + dangling_sum; n];
404        for j in 0..n {
405            let received: f64 = receive_from[j].iter().map(|&(i, w)| pr[i] * w).sum();
406            new_pr[j] += d * received;
407        }
408
409        // Check convergence: L1 norm.
410        let delta: f64 = pr
411            .iter()
412            .zip(new_pr.iter())
413            .map(|(a, b)| (a - b).abs())
414            .sum();
415        pr = new_pr;
416
417        if delta < config.tol {
418            converged = true;
419            break;
420        }
421    }
422
423    // Sort: score desc, key asc on ties.
424    let mut scores: Vec<(String, f64)> = node_keys.into_iter().zip(pr).collect();
425    scores.sort_by(|(ka, sa), (kb, sb)| {
426        sb.partial_cmp(sa)
427            .unwrap_or(std::cmp::Ordering::Equal)
428            .then(ka.cmp(kb))
429    });
430
431    PageRankReport { scores, converged }
432}
433
434// ---------------------------------------------------------------------------
435// Weakly-connected components (WCC)
436// ---------------------------------------------------------------------------
437
438/// Configuration for [`GraphDb::connected_components`].
439#[derive(Debug, Clone, Serialize, Deserialize)]
440#[serde(default)]
441pub struct WccConfig {
442    /// Restrict edges to this type. `None` uses all edge types.
443    pub edge_type: Option<String>,
444    /// Wall-clock budget (milliseconds) for the HTTP server endpoint.
445    pub budget_ms: u64,
446    /// Read this edge property as the edge weight; missing or non-numeric
447    /// values fall back to `1.0`. Only used together with `min_weight` — WCC
448    /// itself is unweighted, but a weighted edge can still be filtered out.
449    pub weight_prop: Option<String>,
450    /// Drop edges whose resolved weight is below this threshold before the
451    /// algorithm runs. Applied whether or not `weight_prop` is set.
452    pub min_weight: Option<f64>,
453}
454
455impl Default for WccConfig {
456    fn default() -> Self {
457        Self {
458            edge_type: None,
459            budget_ms: 5_000,
460            weight_prop: None,
461            min_weight: None,
462        }
463    }
464}
465
466/// Result of [`GraphDb::connected_components`].
467#[derive(Debug, Clone, Serialize, Deserialize)]
468pub struct WccReport {
469    /// Each live node and the key of the smallest member of its component
470    /// (deterministic component identifier).  Sorted by (component_id, key).
471    pub components: Vec<(String, String)>,
472    /// `true` if the time budget fired before all nodes were processed.
473    pub truncated: bool,
474}
475
476/// Union-Find with path compression and union by rank.
477struct UnionFind {
478    parent: Vec<usize>,
479    rank: Vec<u8>,
480}
481
482impl UnionFind {
483    fn new(n: usize) -> Self {
484        Self {
485            parent: (0..n).collect(),
486            rank: vec![0; n],
487        }
488    }
489
490    fn find(&mut self, mut x: usize) -> usize {
491        while self.parent[x] != x {
492            self.parent[x] = self.parent[self.parent[x]]; // path halving
493            x = self.parent[x];
494        }
495        x
496    }
497
498    fn union(&mut self, a: usize, b: usize) {
499        let ra = self.find(a);
500        let rb = self.find(b);
501        if ra == rb {
502            return;
503        }
504        match self.rank[ra].cmp(&self.rank[rb]) {
505            std::cmp::Ordering::Less => self.parent[ra] = rb,
506            std::cmp::Ordering::Greater => self.parent[rb] = ra,
507            std::cmp::Ordering::Equal => {
508                self.parent[rb] = ra;
509                self.rank[ra] += 1;
510            }
511        }
512    }
513}
514
515/// Run weakly-connected components on the unified topology (undirected).
516///
517/// Component ID is the smallest member key in each component (deterministic).
518pub(crate) fn wcc(
519    topo: &TopologyView,
520    idmap: &IdMap,
521    syms: &Interner,
522    labels: &[u32],
523    edge_props: &EdgePropsView,
524    config: &WccConfig,
525) -> WccReport {
526    let weighted = config.weight_prop.is_some() || config.min_weight.is_some();
527    let deadline = if config.budget_ms > 0 {
528        Some(Instant::now() + Duration::from_millis(config.budget_ms))
529    } else {
530        None
531    };
532
533    let (node_ids, node_keys) = live_nodes(idmap, labels);
534    let n = node_ids.len();
535
536    if n == 0 {
537        return WccReport {
538            components: Vec::new(),
539            truncated: false,
540        };
541    }
542
543    // Map internal id → compact index.
544    let mut id_to_idx: BTreeMap<u32, usize> = BTreeMap::new();
545    for (i, &id) in node_ids.iter().enumerate() {
546        id_to_idx.insert(id, i);
547    }
548
549    // Resolve etype filter.
550    let etype_filter = match resolve_etype(syms, config.edge_type.as_deref()) {
551        None => {
552            // No edges of this type → every node is its own component.
553            let mut components: Vec<(String, String)> =
554                node_keys.iter().map(|k| (k.clone(), k.clone())).collect();
555            components.sort();
556            return WccReport {
557                components,
558                truncated: false,
559            };
560        }
561        Some(f) => f,
562    };
563
564    let etypes = etypes_filtered(topo, etype_filter);
565
566    let mut uf = UnionFind::new(n);
567    let mut truncated = false;
568
569    // Union all edges (both directions — WCC treats graph as undirected).
570    'outer: for &et in &etypes {
571        for (i, &id) in node_ids.iter().enumerate() {
572            if let Some(dl) = deadline {
573                if Instant::now() >= dl {
574                    truncated = true;
575                    break 'outer;
576                }
577            }
578            // Out-edges: union i with each out-neighbor.
579            for &nbr in topo.neighbors(et, Direction::Out, id).as_ref() {
580                if let Some(&j) = id_to_idx.get(&nbr) {
581                    if weighted
582                        && edge_weight(
583                            edge_props,
584                            et,
585                            id,
586                            nbr,
587                            config.weight_prop.as_deref(),
588                            config.min_weight,
589                        )
590                        .is_none()
591                    {
592                        continue; // filtered by min_weight
593                    }
594                    uf.union(i, j);
595                }
596            }
597            // In-edges handled by the mirror out-edge from the other side,
598            // but we also cover them here for safety (e.g., self-loops, or
599            // nodes with only in-edges for a filtered etype).
600            for &nbr in topo.neighbors(et, Direction::In, id).as_ref() {
601                if let Some(&j) = id_to_idx.get(&nbr) {
602                    if weighted
603                        && edge_weight(
604                            edge_props,
605                            et,
606                            nbr,
607                            id,
608                            config.weight_prop.as_deref(),
609                            config.min_weight,
610                        )
611                        .is_none()
612                    {
613                        continue; // filtered by min_weight
614                    }
615                    uf.union(i, j);
616                }
617            }
618        }
619    }
620
621    // Determine component representative: smallest key per root.
622    let mut root_min_key: BTreeMap<usize, &str> = BTreeMap::new();
623    for (i, key_str) in node_keys.iter().enumerate() {
624        let root = uf.find(i);
625        let key = key_str.as_str();
626        let entry = root_min_key.entry(root).or_insert(key);
627        if key < *entry {
628            *entry = key;
629        }
630    }
631
632    let mut components: Vec<(String, String)> = node_keys
633        .iter()
634        .enumerate()
635        .map(|(i, key_str)| {
636            let root = uf.find(i);
637            let comp_id = root_min_key[&root].to_string();
638            (key_str.clone(), comp_id)
639        })
640        .collect();
641    components.sort_by(|(ka, ca), (kb, cb)| ca.cmp(cb).then(ka.cmp(kb)));
642
643    WccReport {
644        components,
645        truncated,
646    }
647}
648
649// ---------------------------------------------------------------------------
650// Degree centrality
651// ---------------------------------------------------------------------------
652
653/// Configuration for [`GraphDb::degree_centrality`].
654#[derive(Debug, Clone, Serialize, Deserialize)]
655#[serde(default)]
656pub struct DegreeConfig {
657    /// Restrict edges to this type. `None` counts all edge types.
658    pub edge_type: Option<String>,
659    /// Which edges to count per node.
660    pub direction: AlgoDir,
661    /// Wall-clock budget (milliseconds) for the HTTP server endpoint.
662    pub budget_ms: u64,
663    /// Read this edge property as the edge weight; missing or non-numeric
664    /// values fall back to `1.0`. Degree stays an unweighted count of the
665    /// edges that survive `min_weight` filtering — this does not weight the
666    /// count itself.
667    pub weight_prop: Option<String>,
668    /// Drop edges whose resolved weight is below this threshold before the
669    /// algorithm runs. Applied whether or not `weight_prop` is set.
670    pub min_weight: Option<f64>,
671}
672
673impl Default for DegreeConfig {
674    fn default() -> Self {
675        Self {
676            edge_type: None,
677            direction: AlgoDir::Both,
678            budget_ms: 5_000,
679            weight_prop: None,
680            min_weight: None,
681        }
682    }
683}
684
685/// Result of [`GraphDb::degree_centrality`].
686#[derive(Debug, Clone, Serialize, Deserialize)]
687pub struct DegreeReport {
688    /// Node keys and their degree.  Sorted: degree descending, key ascending on ties.
689    pub scores: Vec<(String, u64)>,
690    /// `true` if the time budget fired before all nodes were processed.
691    pub truncated: bool,
692}
693
694/// Compute degree centrality for all live nodes.
695pub(crate) fn degree_centrality(
696    topo: &TopologyView,
697    idmap: &IdMap,
698    syms: &Interner,
699    labels: &[u32],
700    edge_props: &EdgePropsView,
701    config: &DegreeConfig,
702) -> DegreeReport {
703    let weighted = config.weight_prop.is_some() || config.min_weight.is_some();
704    let deadline = if config.budget_ms > 0 {
705        Some(Instant::now() + Duration::from_millis(config.budget_ms))
706    } else {
707        None
708    };
709
710    let (node_ids, node_keys) = live_nodes(idmap, labels);
711    let n = node_ids.len();
712
713    if n == 0 {
714        return DegreeReport {
715            scores: Vec::new(),
716            truncated: false,
717        };
718    }
719
720    // Resolve etype filter.
721    let etype_filter = match resolve_etype(syms, config.edge_type.as_deref()) {
722        None => {
723            // No edges of this type → all degrees are 0.
724            let scores = node_keys.iter().map(|k| (k.clone(), 0u64)).collect();
725            return DegreeReport {
726                scores,
727                truncated: false,
728            };
729        }
730        Some(f) => f,
731    };
732
733    let etypes = etypes_filtered(topo, etype_filter);
734    let mut degrees: Vec<u64> = vec![0u64; n];
735    let mut truncated = false;
736
737    for (i, &id) in node_ids.iter().enumerate() {
738        if let Some(dl) = deadline {
739            if Instant::now() >= dl {
740                truncated = true;
741                break;
742            }
743        }
744        for &et in &etypes {
745            let dirs: &[Direction] = match config.direction {
746                AlgoDir::Out => &[Direction::Out],
747                AlgoDir::In => &[Direction::In],
748                AlgoDir::Both => &[Direction::Out, Direction::In],
749            };
750            for &dir in dirs {
751                if !weighted {
752                    degrees[i] += topo.neighbors(et, dir, id).len() as u64;
753                    continue;
754                }
755                for &nbr in topo.neighbors(et, dir, id).as_ref() {
756                    let (src, dst) = match dir {
757                        Direction::Out => (id, nbr),
758                        Direction::In => (nbr, id),
759                    };
760                    if edge_weight(
761                        edge_props,
762                        et,
763                        src,
764                        dst,
765                        config.weight_prop.as_deref(),
766                        config.min_weight,
767                    )
768                    .is_some()
769                    {
770                        degrees[i] += 1;
771                    }
772                }
773            }
774        }
775    }
776
777    let mut scores: Vec<(String, u64)> = node_keys.into_iter().zip(degrees).collect();
778    scores.sort_by(|(ka, da), (kb, db)| db.cmp(da).then(ka.cmp(kb)));
779
780    DegreeReport { scores, truncated }
781}
782
783// ---------------------------------------------------------------------------
784// Louvain community detection
785// ---------------------------------------------------------------------------
786
787/// Weighted undirected adjacency list: `adj[i]` is `(neighbor, weight)` pairs
788/// for compact node index `i`. Never contains a self-entry (`i == neighbor`)
789/// — internal/self weight is tracked separately (see `local_moving`,
790/// `aggregate`).
791type WeightedAdj = Vec<Vec<(usize, f64)>>;
792
793/// The next (coarser) level's graph, built by [`aggregate`].
794struct AggregatedLevel {
795    /// `renumbered[i]` is node `i`'s (at the level just aggregated) new,
796    /// compact `0..n` community id — callers compose this directly into
797    /// their own node→community mapping.
798    renumbered: Vec<usize>,
799    n: usize,
800    adj: WeightedAdj,
801    self_weight: Vec<f64>,
802}
803
804/// Configuration for [`GraphDb::communities`].
805#[derive(Debug, Clone, Serialize, Deserialize)]
806#[serde(default)]
807pub struct LouvainConfig {
808    /// Restrict to the union of these edge types. Empty means all edge types
809    /// (manual + rule-derived, via the unified topology).
810    pub edge_types: Vec<String>,
811    /// Read this edge property as the edge weight; missing or non-numeric
812    /// values fall back to `1.0`. `None` treats every edge as weight `1.0`.
813    pub weight_prop: Option<String>,
814    /// Drop edges whose resolved weight is below this threshold before the
815    /// algorithm runs. Applied whether or not `weight_prop` is set.
816    pub min_weight: Option<f64>,
817    /// Modularity resolution parameter (`γ`). Default `1.0`; values above 1
818    /// favor more, smaller communities; below 1 favor fewer, larger ones.
819    pub resolution: f64,
820    /// Maximum number of local-moving + aggregation passes.
821    pub max_passes: u32,
822    /// Maximum local-moving sweeps within a single pass.
823    pub max_sweeps: u32,
824    /// Wall-clock budget (milliseconds), checked once per sweep. `0` means no
825    /// budget (run to convergence or `max_passes`/`max_sweeps`).
826    pub budget_ms: u64,
827    /// Restrict membership to nodes carrying this label. Edges touching a
828    /// node outside the label set are ignored.
829    pub node_label: Option<String>,
830}
831
832impl Default for LouvainConfig {
833    fn default() -> Self {
834        Self {
835            edge_types: Vec::new(),
836            weight_prop: None,
837            min_weight: None,
838            resolution: 1.0,
839            max_passes: 10,
840            max_sweeps: 20,
841            budget_ms: 5_000,
842            node_label: None,
843        }
844    }
845}
846
847/// One detected community.
848#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
849pub struct Community {
850    /// 0-based id assigned by output order (position in
851    /// [`CommunityReport::communities`]) — not stable across different runs
852    /// with a different partition shape.
853    pub id: u32,
854    /// Member node keys, sorted ascending.
855    pub members: Vec<String>,
856    /// Total weight of edges with both endpoints inside this community (each
857    /// undirected edge counted once), computed from the original edges after
858    /// `weight_prop`/`min_weight` filtering — not from the aggregated
859    /// intermediate levels the algorithm builds internally.
860    pub internal_weight: f64,
861    /// `internal_weight / (internal_weight + weight of edges leaving the
862    /// community)`. `1.0` for a community with no incident edges at all
863    /// (trivially cohesive).
864    pub cohesion: f64,
865}
866
867/// Result of [`GraphDb::communities`].
868#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
869pub struct CommunityReport {
870    /// Sorted: size descending, then smallest member key ascending on ties.
871    pub communities: Vec<Community>,
872    /// Modularity of the final partition (resolution-adjusted).
873    pub modularity: f64,
874    /// `true` if the time budget fired before local moving converged.
875    pub truncated: bool,
876}
877
878/// One level's local-moving phase: greedily reassigns each node to the
879/// neighboring community (or itself) that maximizes modularity gain, in
880/// sorted node order, sweeping until stable or `max_sweeps`.
881///
882/// Returns `(community_of, hit_budget)`. `community_of[i]` is a community id
883/// drawn from `0..n` (not necessarily contiguous). `hit_budget` is `true`
884/// when the deadline fired before local moving converged — the returned
885/// assignment is then whatever the sweeps completed before the deadline.
886fn local_moving(
887    n: usize,
888    adj: &WeightedAdj,
889    self_weight: &[f64],
890    resolution: f64,
891    max_sweeps: u32,
892    deadline: Option<Instant>,
893) -> (Vec<usize>, bool) {
894    let k: Vec<f64> = (0..n)
895        .map(|i| adj[i].iter().map(|&(_, w)| w).sum::<f64>() + 2.0 * self_weight[i])
896        .collect();
897    let m: f64 = k.iter().sum::<f64>() / 2.0;
898    let mut community_of: Vec<usize> = (0..n).collect();
899    if m <= 0.0 {
900        return (community_of, false);
901    }
902    let mut tot: Vec<f64> = k.clone();
903
904    for _sweep in 0..max_sweeps {
905        if let Some(dl) = deadline {
906            if Instant::now() >= dl {
907                return (community_of, true);
908            }
909        }
910        let mut improved = false;
911        for i in 0..n {
912            let ci = community_of[i];
913            tot[ci] -= k[i];
914
915            // Weight from i to each neighboring community, keyed by
916            // community id ascending (BTreeMap) so tie-breaking below is
917            // deterministic regardless of adjacency iteration order.
918            let mut neighbor_weights: BTreeMap<usize, f64> = BTreeMap::new();
919            for &(j, w) in &adj[i] {
920                if j == i {
921                    continue; // no self entries stored in adj; defensive
922                }
923                *neighbor_weights.entry(community_of[j]).or_insert(0.0) += w;
924            }
925
926            let gain = |c: usize, w_in: f64| -> f64 {
927                w_in / m - resolution * tot[c] * k[i] / (2.0 * m * m)
928            };
929
930            let mut best_c = ci;
931            let mut best_gain = gain(ci, neighbor_weights.get(&ci).copied().unwrap_or(0.0));
932            for (&c, &w_in) in &neighbor_weights {
933                if c == ci {
934                    continue;
935                }
936                let g = gain(c, w_in);
937                if g > best_gain + 1e-12 {
938                    best_gain = g;
939                    best_c = c;
940                }
941            }
942
943            tot[best_c] += k[i];
944            if best_c != ci {
945                community_of[i] = best_c;
946                improved = true;
947            }
948        }
949        if !improved {
950            break;
951        }
952    }
953
954    (community_of, false)
955}
956
957/// Build the next level's aggregated (super-node) graph from a completed
958/// local-moving assignment. Each distinct community becomes one super-node;
959/// edges within a community fold into its self-weight, edges crossing
960/// communities sum into the new adjacency.
961///
962/// Returns `None` when no coarsening happened (every node kept its own
963/// singleton community) — a local optimum where further aggregation would
964/// have no effect.
965fn aggregate(
966    n: usize,
967    adj: &WeightedAdj,
968    self_weight: &[f64],
969    community_of: &[usize],
970) -> Option<AggregatedLevel> {
971    let mut remap: BTreeMap<usize, usize> = BTreeMap::new();
972    let mut next_id = 0usize;
973    let mut renumbered: Vec<usize> = vec![0; n];
974    for (i, item) in renumbered.iter_mut().enumerate() {
975        let c = community_of[i];
976        let idx = *remap.entry(c).or_insert_with(|| {
977            let id = next_id;
978            next_id += 1;
979            id
980        });
981        *item = idx;
982    }
983    let new_n = next_id;
984    if new_n == n {
985        return None; // every community is a singleton: no coarsening
986    }
987
988    let mut new_self_weight = vec![0.0; new_n];
989    let mut new_adj_map: Vec<BTreeMap<usize, f64>> = vec![BTreeMap::new(); new_n];
990    for i in 0..n {
991        let ci = renumbered[i];
992        new_self_weight[ci] += self_weight[i];
993        for &(j, w) in &adj[i] {
994            if j < i {
995                continue; // adjacency is symmetric; process each edge once
996            }
997            let cj = renumbered[j];
998            if ci == cj {
999                new_self_weight[ci] += w;
1000            } else {
1001                *new_adj_map[ci].entry(cj).or_insert(0.0) += w;
1002                *new_adj_map[cj].entry(ci).or_insert(0.0) += w;
1003            }
1004        }
1005    }
1006
1007    let new_adj: WeightedAdj = new_adj_map
1008        .into_iter()
1009        .map(|map| map.into_iter().collect())
1010        .collect();
1011
1012    Some(AggregatedLevel {
1013        renumbered,
1014        n: new_n,
1015        adj: new_adj,
1016        self_weight: new_self_weight,
1017    })
1018}
1019
1020/// Run Louvain community detection on the unified topology (undirected).
1021///
1022/// Sums both edge directions into a single undirected weight, ignores
1023/// self-loops, and restricts membership to `config.node_label` when set
1024/// (edges touching a node outside the label set are ignored entirely).
1025/// `cohesion` and `internal_weight` are computed from the original filtered
1026/// edges, not the internal aggregated levels.
1027pub(crate) fn louvain(
1028    topo: &TopologyView,
1029    idmap: &IdMap,
1030    syms: &Interner,
1031    labels: &[u32],
1032    edge_props: &EdgePropsView,
1033    config: &LouvainConfig,
1034) -> CommunityReport {
1035    let deadline = if config.budget_ms > 0 {
1036        Some(Instant::now() + Duration::from_millis(config.budget_ms))
1037    } else {
1038        None
1039    };
1040
1041    // Node set (optionally label-restricted), sorted by key ascending so
1042    // compact index 0 is always the smallest key — local moving then
1043    // processes nodes in sorted key order at level 0, and that order
1044    // propagates deterministically into every aggregated level.
1045    let (raw_ids, raw_keys) =
1046        live_nodes_for_label(idmap, syms, labels, config.node_label.as_deref());
1047    let mut order: Vec<usize> = (0..raw_ids.len()).collect();
1048    order.sort_by(|&a, &b| raw_keys[a].cmp(&raw_keys[b]));
1049    let node_ids: Vec<u32> = order.iter().map(|&i| raw_ids[i]).collect();
1050    let node_keys: Vec<String> = order.iter().map(|&i| raw_keys[i].clone()).collect();
1051    let n0 = node_ids.len();
1052
1053    if n0 == 0 {
1054        return CommunityReport {
1055            communities: Vec::new(),
1056            modularity: 0.0,
1057            truncated: false,
1058        };
1059    }
1060
1061    let mut id_to_idx: BTreeMap<u32, usize> = BTreeMap::new();
1062    for (i, &id) in node_ids.iter().enumerate() {
1063        id_to_idx.insert(id, i);
1064    }
1065
1066    let etypes = resolve_etypes_multi(syms, topo, &config.edge_types);
1067
1068    // Collect the original (filtered) undirected weighted edges: sum both
1069    // directions into one entry per unordered compact-index pair, ignore
1070    // self-loops, ignore edges touching a node outside the label set.
1071    // Keyed by (a, b) with a < b so iteration order (and therefore floating
1072    // point summation order) is deterministic.
1073    let mut edge_weight_map: BTreeMap<(usize, usize), f64> = BTreeMap::new();
1074    for &et in &etypes {
1075        for (i, &id) in node_ids.iter().enumerate() {
1076            for &nbr in topo.neighbors(et, Direction::Out, id).as_ref() {
1077                if nbr == id {
1078                    continue; // ignore self-loops
1079                }
1080                let Some(&j) = id_to_idx.get(&nbr) else {
1081                    continue; // touches a node outside the label restriction
1082                };
1083                let Some(w) = edge_weight(
1084                    edge_props,
1085                    et,
1086                    id,
1087                    nbr,
1088                    config.weight_prop.as_deref(),
1089                    config.min_weight,
1090                ) else {
1091                    continue; // filtered by min_weight
1092                };
1093                let key = if i < j { (i, j) } else { (j, i) };
1094                *edge_weight_map.entry(key).or_insert(0.0) += w;
1095            }
1096        }
1097    }
1098
1099    let m: f64 = edge_weight_map.values().sum();
1100
1101    let mut adj: WeightedAdj = vec![Vec::new(); n0];
1102    for (&(a, b), &w) in &edge_weight_map {
1103        adj[a].push((b, w));
1104        adj[b].push((a, w));
1105    }
1106    let mut self_weight: Vec<f64> = vec![0.0; n0];
1107
1108    // owner[i] = original node i's community index at the current level.
1109    let mut owner: Vec<usize> = (0..n0).collect();
1110    let mut truncated = false;
1111    let mut n = n0;
1112
1113    if m > 0.0 {
1114        'passes: for _pass in 0..config.max_passes {
1115            let (community_of, hit_budget) = local_moving(
1116                n,
1117                &adj,
1118                &self_weight,
1119                config.resolution,
1120                config.max_sweeps,
1121                deadline,
1122            );
1123            if hit_budget {
1124                // Fold this (possibly partial) sweep's assignment straight
1125                // into owner and stop — no further aggregation, so no
1126                // renumbering is needed.
1127                owner = owner.iter().map(|&o| community_of[o]).collect();
1128                truncated = true;
1129                break 'passes;
1130            }
1131            let Some(level) = aggregate(n, &adj, &self_weight, &community_of) else {
1132                // Local optimum: no further coarsening. community_of is
1133                // already the final assignment at this level.
1134                owner = owner.iter().map(|&o| community_of[o]).collect();
1135                break 'passes;
1136            };
1137            // Compose owner directly through the new level's compact
1138            // numbering (folds community_of + remap in one step).
1139            owner = owner.iter().map(|&o| level.renumbered[o]).collect();
1140            n = level.n;
1141            adj = level.adj;
1142            self_weight = level.self_weight;
1143        }
1144    }
1145
1146    // Cohesion / modularity from the ORIGINAL filtered edges, grouped by
1147    // final community.
1148    let mut internal: BTreeMap<usize, f64> = BTreeMap::new();
1149    let mut leaving: BTreeMap<usize, f64> = BTreeMap::new();
1150    for (&(a, b), &w) in &edge_weight_map {
1151        let ca = owner[a];
1152        let cb = owner[b];
1153        if ca == cb {
1154            *internal.entry(ca).or_insert(0.0) += w;
1155        } else {
1156            *leaving.entry(ca).or_insert(0.0) += w;
1157            *leaving.entry(cb).or_insert(0.0) += w;
1158        }
1159    }
1160
1161    let mut members_by_community: BTreeMap<usize, Vec<String>> = BTreeMap::new();
1162    for (i, key) in node_keys.iter().enumerate() {
1163        members_by_community
1164            .entry(owner[i])
1165            .or_default()
1166            .push(key.clone());
1167    }
1168
1169    let modularity = if m > 0.0 {
1170        members_by_community
1171            .keys()
1172            .map(|c| {
1173                let internal_w = internal.get(c).copied().unwrap_or(0.0);
1174                let leaving_w = leaving.get(c).copied().unwrap_or(0.0);
1175                let sigma_tot = 2.0 * internal_w + leaving_w;
1176                internal_w / m - config.resolution * (sigma_tot * sigma_tot) / (4.0 * m * m)
1177            })
1178            .sum()
1179    } else {
1180        0.0
1181    };
1182
1183    let mut communities: Vec<Community> = members_by_community
1184        .into_iter()
1185        .map(|(c, mut members)| {
1186            members.sort();
1187            let internal_w = internal.get(&c).copied().unwrap_or(0.0);
1188            let leaving_w = leaving.get(&c).copied().unwrap_or(0.0);
1189            let cohesion = if internal_w + leaving_w > 0.0 {
1190                internal_w / (internal_w + leaving_w)
1191            } else {
1192                1.0
1193            };
1194            Community {
1195                id: 0, // assigned below, after sorting
1196                members,
1197                internal_weight: internal_w,
1198                cohesion,
1199            }
1200        })
1201        .collect();
1202
1203    communities.sort_by(|a, b| {
1204        b.members
1205            .len()
1206            .cmp(&a.members.len())
1207            .then_with(|| a.members[0].cmp(&b.members[0]))
1208    });
1209    for (i, c) in communities.iter_mut().enumerate() {
1210        c.id = i as u32;
1211    }
1212
1213    CommunityReport {
1214        communities,
1215        modularity,
1216        truncated,
1217    }
1218}