1use super::{
2 direction_sort_rank, normalize_symmetric_direction, DirectedNeighborHit, Direction,
3 EdgeRelation, Entity, FromStr, GraphPath, HashMap, KhiveRuntime, NamespaceToken,
4 NeighborCursor, NeighborHit, NeighborQuery, Note, RuntimeError, RuntimeResult, SqlStatement,
5 SqlValue, TraversalRequest, Uuid,
6};
7
8fn note_graph_name(note: &Note) -> String {
9 note.name
10 .as_deref()
11 .filter(|name| !name.trim().is_empty())
12 .map(str::to_owned)
13 .unwrap_or_else(|| format!("[{}]", note.kind))
14}
15
16pub(super) fn merge_traversal_paths_by_root(
21 paths: Vec<GraphPath>,
22 limit: Option<u32>,
23) -> Vec<GraphPath> {
24 let mut order: Vec<Uuid> = Vec::new();
25 let mut merged: HashMap<Uuid, GraphPath> = HashMap::new();
26 let mut node_index: HashMap<Uuid, HashMap<Uuid, usize>> = HashMap::new();
30
31 for path in paths {
32 let existing = merged.entry(path.root_id).or_insert_with(|| {
33 order.push(path.root_id);
34 GraphPath {
35 root_id: path.root_id,
36 nodes: Vec::new(),
37 total_weight: 0.0,
38 }
39 });
40 let index = node_index.entry(path.root_id).or_default();
41 for node in path.nodes {
42 match index.get(&node.node_id) {
43 Some(&i) => {
44 if node.depth < existing.nodes[i].depth {
45 existing.nodes[i] = node;
46 }
47 }
48 None => {
49 index.insert(node.node_id, existing.nodes.len());
50 existing.nodes.push(node);
51 }
52 }
53 }
54 }
55
56 order
57 .into_iter()
58 .filter_map(|root_id| merged.remove(&root_id))
59 .map(|mut path| {
60 path.nodes.sort_by_key(|n| n.depth);
62 if let Some(lim) = limit {
63 let lim = lim as usize;
64 let mut non_root_kept = 0usize;
65 path.nodes.retain(|n| {
66 if n.depth == 0 {
67 return true;
68 }
69 if non_root_kept < lim {
70 non_root_kept += 1;
71 true
72 } else {
73 false
74 }
75 });
76 }
77 recompute_total_weight(&mut path);
78 path
79 })
80 .collect()
81}
82
83fn recompute_total_weight(path: &mut GraphPath) {
92 path.total_weight = path.nodes.iter().map(|n| n.weight).fold(0.0_f64, f64::max);
93}
94
95impl KhiveRuntime {
96 pub(crate) async fn substrate_exists_in_ns(
103 &self,
104 token: &NamespaceToken,
105 id: Uuid,
106 ) -> RuntimeResult<bool> {
107 if self.resolve(token, id).await?.is_some() {
108 return Ok(true);
109 }
110 match self.get_edge_visible(token, id).await {
111 Ok(Some(_)) => Ok(true),
112 Ok(None) | Err(RuntimeError::NotFound(_)) => Ok(false),
113 Err(err) => Err(err),
114 }
115 }
116
117 pub(crate) async fn substrate_exists_by_id(
124 &self,
125 token: &NamespaceToken,
126 id: Uuid,
127 ) -> RuntimeResult<bool> {
128 if self.resolve_edge_endpoint(token, id).await?.is_some() {
129 return Ok(true);
130 }
131 match self.get_edge(token, id).await {
132 Ok(Some(_)) => Ok(true),
133 Ok(None) | Err(RuntimeError::NotFound(_)) => Ok(false),
134 Err(err) => Err(err),
135 }
136 }
137
138 pub async fn latest_annotating_note(
143 &self,
144 token: &NamespaceToken,
145 node_id: Uuid,
146 kind: &str,
147 tag: &str,
148 ) -> RuntimeResult<Option<Uuid>> {
149 self.latest_annotating_note_inner(token, node_id, kind, tag, None)
150 .await
151 }
152
153 pub async fn latest_annotating_note_with_property(
156 &self,
157 token: &NamespaceToken,
158 node_id: Uuid,
159 kind: &str,
160 tag: &str,
161 property_key: &str,
162 property_value: &str,
163 ) -> RuntimeResult<Option<Uuid>> {
164 self.latest_annotating_note_inner(
165 token,
166 node_id,
167 kind,
168 tag,
169 Some((property_key, property_value)),
170 )
171 .await
172 }
173
174 async fn latest_annotating_note_inner(
175 &self,
176 token: &NamespaceToken,
177 node_id: Uuid,
178 kind: &str,
179 tag: &str,
180 required_property: Option<(&str, &str)>,
181 ) -> RuntimeResult<Option<Uuid>> {
182 if !self.substrate_exists_in_ns(token, node_id).await? {
183 return Ok(None);
184 }
185 let mut latest: Option<(Uuid, i64)> = None;
186 for namespace in token.visible_namespaces() {
187 let scoped = NamespaceToken::for_namespace(namespace.clone());
188 let graph = self.graph(&scoped)?;
189 let candidate = match required_property {
190 Some((key, value)) => {
191 graph
192 .latest_annotating_note_with_property(node_id, kind, tag, key, value)
193 .await?
194 }
195 None => graph.latest_annotating_note(node_id, kind, tag).await?,
196 };
197 if let Some(candidate) = candidate {
198 if latest.is_none_or(|(id, created_at)| {
199 candidate.1 > created_at || (candidate.1 == created_at && candidate.0 < id)
200 }) {
201 latest = Some(candidate);
202 }
203 }
204 }
205 Ok(latest.map(|(id, _)| id))
206 }
207
208 pub async fn neighbors(
217 &self,
218 token: &NamespaceToken,
219 node_id: Uuid,
220 direction: Direction,
221 limit: Option<u32>,
222 relations: Option<Vec<EdgeRelation>>,
223 ) -> RuntimeResult<Vec<NeighborHit>> {
224 self.neighbors_with_query(
225 token,
226 node_id,
227 NeighborQuery {
228 direction,
229 relations,
230 limit,
231 min_weight: None,
232 },
233 )
234 .await
235 }
236
237 pub async fn neighbors_with_query(
247 &self,
248 token: &NamespaceToken,
249 node_id: Uuid,
250 query: NeighborQuery,
251 ) -> RuntimeResult<Vec<NeighborHit>> {
252 self.neighbors_with_query_page(token, node_id, query, None, None, true)
253 .await
254 }
255
256 pub async fn neighbors_with_query_page(
260 &self,
261 token: &NamespaceToken,
262 node_id: Uuid,
263 query: NeighborQuery,
264 after: Option<NeighborCursor>,
265 neighbor_kinds: Option<Vec<String>>,
266 enrich: bool,
267 ) -> RuntimeResult<Vec<NeighborHit>> {
268 if !self.substrate_exists_by_id(token, node_id).await? {
271 return Err(RuntimeError::NotFound(format!(
272 "neighbor anchor {node_id} not found"
273 )));
274 }
275
276 self.neighbors_for_resolved_kg_read(
277 token,
278 node_id,
279 crate::KgNeighborRead {
280 query,
281 after,
282 neighbor_kinds,
283 enrich,
284 namespace: None,
285 },
286 )
287 .await
288 }
289
290 pub async fn neighbors_for_resolved_kg_read(
298 &self,
299 token: &NamespaceToken,
300 node_id: Uuid,
301 options: crate::KgNeighborRead,
302 ) -> RuntimeResult<Vec<NeighborHit>> {
303 self.neighbors_for_resolved_kg_read_inner(token, node_id, options, false)
304 .await
305 .map(|(hits, _)| hits)
306 }
307
308 pub async fn neighbors_for_resolved_kg_read_with_entity_kinds(
313 &self,
314 token: &NamespaceToken,
315 node_id: Uuid,
316 options: crate::KgNeighborRead,
317 ) -> RuntimeResult<(Vec<NeighborHit>, HashMap<Uuid, String>)> {
318 self.neighbors_for_resolved_kg_read_inner(token, node_id, options, true)
319 .await
320 }
321
322 async fn neighbors_for_resolved_kg_read_inner(
323 &self,
324 token: &NamespaceToken,
325 node_id: Uuid,
326 options: crate::KgNeighborRead,
327 with_entity_kinds: bool,
328 ) -> RuntimeResult<(Vec<NeighborHit>, HashMap<Uuid, String>)> {
329 let crate::KgNeighborRead {
330 mut query,
331 after,
332 neighbor_kinds,
333 enrich,
334 namespace,
335 } = options;
336 let namespaces = crate::kg_read::neighbor_read_namespaces(token, namespace.as_ref())?;
337 query.direction =
338 normalize_symmetric_direction(query.direction, query.relations.as_deref());
339 let mut hits = Vec::new();
340 for ns in namespaces {
341 let temp = NamespaceToken::for_namespace(ns.clone());
342 let mut ns_hits = self
343 .graph(&temp)?
344 .neighbors_page(node_id, query.clone(), after, neighbor_kinds.clone())
345 .await?;
346 hits.append(&mut ns_hits);
347 }
348 hits.sort_by_key(|h| (h.node_id, h.edge_id));
349 hits.dedup_by_key(|h| (h.node_id, h.edge_id));
350 if enrich {
351 self.enrich_neighbor_hits(token, &mut hits).await;
352 }
353 let candidate_ids: Vec<Uuid> = hits.iter().map(|h| h.node_id).collect();
355 let (deleted, entity_kinds) = self
356 .neighbor_node_screen(
357 candidate_ids,
358 (with_entity_kinds && !enrich).then_some(token),
359 )
360 .await?;
361 if !deleted.is_empty() {
362 hits.retain(|h| !deleted.contains(&h.node_id));
363 }
364 hits.sort_by(|a, b| {
371 b.weight
372 .partial_cmp(&a.weight)
373 .unwrap_or(std::cmp::Ordering::Equal)
374 .then(a.node_id.cmp(&b.node_id))
375 .then(a.edge_id.cmp(&b.edge_id))
376 });
377 Ok((hits, entity_kinds))
378 }
379
380 pub async fn annotation_neighbors_by_target_id(
388 &self,
389 target_id: Uuid,
390 ) -> RuntimeResult<Vec<NeighborHit>> {
391 let mut reader = self.sql().reader().await?;
392 let rows = reader
393 .query_all(SqlStatement {
394 sql: "SELECT source_id, id, weight FROM graph_edges \
395 WHERE target_id = ?1 AND relation = ?2 AND deleted_at IS NULL \
396 ORDER BY weight DESC, source_id ASC"
397 .to_string(),
398 params: vec![
399 SqlValue::Text(target_id.to_string()),
400 SqlValue::Text(EdgeRelation::Annotates.to_string()),
401 ],
402 label: Some("annotations.by_target_id_unfiltered".into()),
403 })
404 .await?;
405
406 rows.into_iter()
407 .map(|row| {
408 let parse_uuid = |name: &str| match row.get(name) {
409 Some(SqlValue::Text(value)) => Uuid::from_str(value).map_err(|error| {
410 RuntimeError::Internal(format!("graph_edges.{name} is not a UUID: {error}"))
411 }),
412 Some(value) => Err(RuntimeError::Internal(format!(
413 "graph_edges.{name} has unexpected SQL value {value:?}"
414 ))),
415 None => Err(RuntimeError::Internal(format!(
416 "graph_edges row missing {name}"
417 ))),
418 };
419 let weight = match row.get("weight") {
420 Some(SqlValue::Float(value)) => Ok(*value),
421 Some(value) => Err(RuntimeError::Internal(format!(
422 "graph_edges.weight has unexpected SQL value {value:?}"
423 ))),
424 None => Err(RuntimeError::Internal(
425 "graph_edges row missing weight".into(),
426 )),
427 }?;
428
429 Ok(NeighborHit {
430 node_id: parse_uuid("source_id")?,
431 edge_id: parse_uuid("id")?,
432 relation: EdgeRelation::Annotates,
433 weight,
434 name: None,
435 kind: None,
436 entity_type: None,
437 })
438 })
439 .collect()
440 }
441
442 pub async fn neighbors_with_query_directed(
451 &self,
452 token: &NamespaceToken,
453 node_id: Uuid,
454 query: NeighborQuery,
455 ) -> RuntimeResult<Vec<(NeighborHit, Direction)>> {
456 if !self.substrate_exists_by_id(token, node_id).await? {
457 return Err(RuntimeError::NotFound(format!(
458 "neighbor anchor {node_id} not found"
459 )));
460 }
461
462 self.directed_neighbors_for_resolved_kg_read(token, node_id, query, None)
463 .await
464 }
465
466 pub(crate) async fn directed_neighbors_for_resolved_kg_read(
467 &self,
468 token: &NamespaceToken,
469 node_id: Uuid,
470 query: NeighborQuery,
471 namespace: Option<&crate::Namespace>,
472 ) -> RuntimeResult<Vec<(NeighborHit, Direction)>> {
473 let namespaces = crate::kg_read::neighbor_read_namespaces(token, namespace)?;
474 let mut hits: Vec<DirectedNeighborHit> = Vec::new();
475 for ns in namespaces {
476 let temp = NamespaceToken::for_namespace(ns.clone());
477 let mut ns_hits = self
478 .graph(&temp)?
479 .neighbors_both_directions(node_id, query.clone())
480 .await?;
481 hits.append(&mut ns_hits);
482 }
483 hits.sort_by_key(|h| {
487 (
488 h.hit.node_id,
489 h.hit.edge_id,
490 direction_sort_rank(&h.direction),
491 )
492 });
493 hits.dedup_by_key(|h| {
494 (
495 h.hit.node_id,
496 h.hit.edge_id,
497 direction_sort_rank(&h.direction),
498 )
499 });
500
501 let mut plain_hits: Vec<NeighborHit> = hits.iter().map(|h| h.hit.clone()).collect();
502 self.enrich_neighbor_hits(token, &mut plain_hits).await;
503 for (dh, enriched) in hits.iter_mut().zip(plain_hits) {
504 dh.hit = enriched;
505 }
506
507 let candidate_ids: Vec<Uuid> = hits.iter().map(|h| h.hit.node_id).collect();
509 let deleted = self.deleted_entity_ids(candidate_ids).await?;
510 if !deleted.is_empty() {
511 hits.retain(|h| !deleted.contains(&h.hit.node_id));
512 }
513 hits.sort_by(|a, b| {
517 b.hit
518 .weight
519 .partial_cmp(&a.hit.weight)
520 .unwrap_or(std::cmp::Ordering::Equal)
521 .then(a.hit.node_id.cmp(&b.hit.node_id))
522 .then(a.hit.edge_id.cmp(&b.hit.edge_id))
523 });
524 Ok(hits.into_iter().map(|h| (h.hit, h.direction)).collect())
525 }
526
527 pub async fn traverse(
533 &self,
534 token: &NamespaceToken,
535 request: TraversalRequest,
536 ) -> RuntimeResult<Vec<GraphPath>> {
537 let mut request = request;
538 request.validate().map_err(RuntimeError::InvalidInput)?;
539 let mut roots = Vec::with_capacity(request.roots.len());
540 let mut seen_roots = std::collections::HashSet::with_capacity(request.roots.len());
541 for root in request.roots.drain(..) {
542 if seen_roots.insert(root) {
543 if !self.substrate_exists_by_id(token, root).await? {
544 return Err(RuntimeError::NotFound(format!(
545 "traverse root {root} not found"
546 )));
547 }
548 roots.push(root);
549 }
550 }
551 request.roots = roots;
552 if request.roots.is_empty() {
553 return Ok(Vec::new());
554 }
555
556 let mut paths = Vec::new();
557 for ns in token.visible_namespaces() {
558 let temp = NamespaceToken::for_namespace(ns.clone());
559 let mut ns_paths = self.graph(&temp)?.traverse(request.clone()).await?;
560 paths.append(&mut ns_paths);
561 }
562 let mut paths =
566 merge_traversal_paths_by_root(paths, Some(request.options.effective_limit()));
567 self.enrich_path_nodes(token, &mut paths, request.include_properties)
568 .await;
569 let all_node_ids: Vec<Uuid> = paths
571 .iter()
572 .flat_map(|p| p.nodes.iter().map(|n| n.node_id))
573 .collect();
574 let deleted = self.deleted_entity_ids(all_node_ids).await?;
575 if !deleted.is_empty() {
576 for path in paths.iter_mut() {
577 path.nodes.retain(|n| !deleted.contains(&n.node_id));
578 recompute_total_weight(path);
579 }
580 paths.retain(|p| !p.nodes.is_empty());
581 }
582 Ok(paths)
583 }
584
585 pub(super) async fn deleted_entity_ids(
605 &self,
606 ids: Vec<Uuid>,
607 ) -> RuntimeResult<std::collections::HashSet<Uuid>> {
608 self.neighbor_node_screen(ids, None)
609 .await
610 .map(|(deleted, _)| deleted)
611 }
612
613 async fn neighbor_node_screen(
617 &self,
618 ids: Vec<Uuid>,
619 kind_token: Option<&NamespaceToken>,
620 ) -> RuntimeResult<(std::collections::HashSet<Uuid>, HashMap<Uuid, String>)> {
621 if ids.is_empty() {
622 return Ok((std::collections::HashSet::new(), HashMap::new()));
623 }
624 let id_strs: Vec<String> = ids.iter().map(|u| u.to_string()).collect();
625 let n = id_strs.len();
626 let entities_placeholders = (0..n)
633 .map(|i| format!("?{}", i + 1))
634 .collect::<Vec<_>>()
635 .join(",");
636 let notes_placeholders = (0..n)
637 .map(|i| format!("?{}", n + i + 1))
638 .collect::<Vec<_>>()
639 .join(",");
640 let sql_str = if kind_token.is_some() {
641 format!(
642 "SELECT id, kind, namespace, deleted_at IS NOT NULL AS is_deleted \
643 FROM entities WHERE id IN ({entities_placeholders}) \
644 UNION ALL \
645 SELECT id, NULL, NULL, 1 FROM notes \
646 WHERE id IN ({notes_placeholders}) AND deleted_at IS NOT NULL"
647 )
648 } else {
649 format!(
650 "SELECT id FROM entities WHERE id IN ({entities_placeholders}) AND deleted_at IS NOT NULL \
651 UNION \
652 SELECT id FROM notes WHERE id IN ({notes_placeholders}) AND deleted_at IS NOT NULL"
653 )
654 };
655 let params: Vec<SqlValue> = id_strs
657 .iter()
658 .chain(id_strs.iter())
659 .cloned()
660 .map(SqlValue::Text)
661 .collect();
662 let stmt = SqlStatement {
663 sql: sql_str,
664 params,
665 label: Some("deleted_entity_ids".into()),
666 };
667 let mut out = std::collections::HashSet::new();
668 let mut entity_kinds = HashMap::new();
669 let sql = self.sql();
670 let mut reader = sql.reader().await?;
671 let rows = reader.query_all(stmt).await?;
672 for row in rows {
673 if let Some(col) = row.columns.first() {
674 if let SqlValue::Text(s) = &col.value {
675 if let Ok(u) = s.parse::<Uuid>() {
676 if kind_token.is_none()
677 || matches!(
678 row.columns.get(3).map(|col| &col.value),
679 Some(SqlValue::Integer(1))
680 )
681 {
682 out.insert(u);
683 } else if let (
684 Some(token),
685 Some(SqlValue::Text(kind)),
686 Some(SqlValue::Text(namespace)),
687 Some(SqlValue::Integer(0)),
688 ) = (
689 kind_token,
690 row.columns.get(1).map(|col| &col.value),
691 row.columns.get(2).map(|col| &col.value),
692 row.columns.get(3).map(|col| &col.value),
693 ) {
694 if token
695 .visible_namespaces()
696 .iter()
697 .any(|ns| ns.as_str() == namespace.as_str())
698 {
699 entity_kinds.insert(u, kind.clone());
700 }
701 }
702 }
703 }
704 }
705 }
706 Ok((out, entity_kinds))
707 }
708
709 pub(super) async fn enrich_neighbor_hits(
718 &self,
719 token: &NamespaceToken,
720 hits: &mut [NeighborHit],
721 ) {
722 if hits.is_empty() {
723 return;
724 }
725
726 let unique_ids: Vec<Uuid> = {
728 let mut seen = std::collections::HashSet::new();
729 hits.iter()
730 .filter_map(|h| {
731 if seen.insert(h.node_id) {
732 Some(h.node_id)
733 } else {
734 None
735 }
736 })
737 .collect()
738 };
739
740 let entity_map: HashMap<Uuid, Entity> = self
741 .get_entities_by_ids_visible(token, &unique_ids)
742 .await
743 .unwrap_or_default()
744 .into_iter()
745 .map(|e| (e.id, e))
746 .collect();
747
748 let residual_ids: Vec<Uuid> = unique_ids
750 .iter()
751 .filter(|id| !entity_map.contains_key(id))
752 .copied()
753 .collect();
754
755 let note_map: HashMap<Uuid, Note> = if !residual_ids.is_empty() {
756 if let Ok(store) = self.notes(token) {
757 store
758 .get_notes_batch(&residual_ids)
759 .await
760 .unwrap_or_default()
761 .into_iter()
762 .map(|n| (n.id, n))
763 .collect()
764 } else {
765 HashMap::new()
766 }
767 } else {
768 HashMap::new()
769 };
770
771 for hit in hits.iter_mut() {
772 if let Some(entity) = entity_map.get(&hit.node_id) {
773 hit.name = Some(entity.name.clone());
774 hit.kind = Some(entity.kind.clone());
775 hit.entity_type = entity.entity_type.clone();
776 } else if let Some(note) = note_map.get(&hit.node_id) {
777 hit.name = Some(note_graph_name(note));
778 hit.kind = Some(note.kind.clone());
779 }
780 }
781 }
782
783 pub(super) async fn enrich_path_nodes(
794 &self,
795 token: &NamespaceToken,
796 paths: &mut [GraphPath],
797 include_properties: bool,
798 ) {
799 if paths.is_empty() {
800 return;
801 }
802
803 let unique_ids: Vec<Uuid> = {
805 let mut seen = std::collections::HashSet::new();
806 paths
807 .iter()
808 .flat_map(|p| p.nodes.iter())
809 .filter_map(|n| {
810 if seen.insert(n.node_id) {
811 Some(n.node_id)
812 } else {
813 None
814 }
815 })
816 .collect()
817 };
818
819 let entity_map: HashMap<Uuid, Entity> = self
820 .get_entities_by_ids_visible(token, &unique_ids)
821 .await
822 .unwrap_or_default()
823 .into_iter()
824 .map(|e| (e.id, e))
825 .collect();
826
827 let residual_ids: Vec<Uuid> = unique_ids
828 .iter()
829 .filter(|id| !entity_map.contains_key(id))
830 .copied()
831 .collect();
832
833 let note_map: HashMap<Uuid, Note> = if !residual_ids.is_empty() {
834 if let Ok(store) = self.notes(token) {
835 store
836 .get_notes_batch(&residual_ids)
837 .await
838 .unwrap_or_default()
839 .into_iter()
840 .map(|n| (n.id, n))
841 .collect()
842 } else {
843 HashMap::new()
844 }
845 } else {
846 HashMap::new()
847 };
848
849 for path in paths.iter_mut() {
850 for node in path.nodes.iter_mut() {
851 if let Some(entity) = entity_map.get(&node.node_id) {
852 node.name = Some(entity.name.clone());
853 node.kind = Some(entity.kind.clone());
854 if include_properties {
855 node.properties = entity.properties.clone();
856 }
857 } else if let Some(note) = note_map.get(&node.node_id) {
858 node.name = Some(note_graph_name(note));
859 node.kind = Some(note.kind.clone());
860 }
861 }
862 }
863 }
864}