1use anyhow::{Context, Result};
2use std::collections::{BTreeMap, BTreeSet};
3use std::path::Path;
4use tsift_core::{GraphEdge, GraphNode, GraphPath, GraphStore, SQLITE_GRAPH_SCHEMA_VERSION};
5
6fn block_on<F: std::future::Future>(rt: &tokio::runtime::Runtime, f: F) -> F::Output {
7 rt.block_on(f)
8}
9
10pub struct LibsqlGraphStore {
11 conn: libsql::Connection,
12 rt: tokio::runtime::Runtime,
13}
14
15impl LibsqlGraphStore {
16 pub fn open(db_path: &Path) -> Result<Self> {
17 if let Some(parent) = db_path.parent() {
18 std::fs::create_dir_all(parent).with_context(|| {
19 format!("creating libsql graph substrate dir: {}", parent.display())
20 })?;
21 }
22 let rt = tokio::runtime::Runtime::new().context("creating tokio runtime for libsql")?;
23 let db = block_on(&rt, libsql::Builder::new_local(db_path).build())
24 .with_context(|| format!("opening libsql graph substrate db: {}", db_path.display()))?;
25 let conn = db.connect().with_context(|| {
26 format!(
27 "connecting to libsql graph substrate db: {}",
28 db_path.display()
29 )
30 })?;
31 let store = Self { conn, rt };
32 store.init_schema()?;
33 Ok(store)
34 }
35
36 pub fn open_remote(url: &str, auth_token: &str) -> Result<Self> {
37 let rt =
38 tokio::runtime::Runtime::new().context("creating tokio runtime for libsql remote")?;
39 let db = block_on(
40 &rt,
41 libsql::Builder::new_remote(url.to_string(), auth_token.to_string()).build(),
42 )
43 .context("building libsql remote database")?;
44 let conn = db
45 .connect()
46 .context("connecting to libsql remote database")?;
47 let store = Self { conn, rt };
48 store.init_schema()?;
49 Ok(store)
50 }
51
52 pub fn in_memory() -> Result<Self> {
53 let rt = tokio::runtime::Runtime::new().context("creating tokio runtime for libsql")?;
54 let db = block_on(&rt, libsql::Builder::new_local(":memory:").build())
55 .context("opening in-memory libsql database")?;
56 let conn = db
57 .connect()
58 .context("connecting to in-memory libsql database")?;
59 let store = Self { conn, rt };
60 store.init_schema()?;
61 Ok(store)
62 }
63
64 fn init_schema(&self) -> Result<()> {
65 block_on(&self.rt, async {
66 self.conn
67 .execute_batch(&format!(
68 r#"
69 PRAGMA foreign_keys = ON;
70 PRAGMA busy_timeout = 5000;
71
72 CREATE TABLE IF NOT EXISTS graph_nodes (
73 id TEXT PRIMARY KEY,
74 kind TEXT NOT NULL,
75 label TEXT NOT NULL,
76 properties_json TEXT NOT NULL DEFAULT '{{}}',
77 provenance_json TEXT NOT NULL DEFAULT '[]',
78 freshness_json TEXT,
79 row_hash TEXT,
80 source_watermark TEXT
81 );
82 CREATE INDEX IF NOT EXISTS idx_graph_nodes_kind
83 ON graph_nodes(kind);
84 CREATE INDEX IF NOT EXISTS idx_graph_nodes_kind_label
85 ON graph_nodes(kind, label, id);
86
87 CREATE TABLE IF NOT EXISTS graph_edges (
88 edge_key TEXT NOT NULL UNIQUE,
89 from_id TEXT NOT NULL,
90 to_id TEXT NOT NULL,
91 kind TEXT NOT NULL,
92 properties_json TEXT NOT NULL DEFAULT '{{}}',
93 provenance_json TEXT NOT NULL DEFAULT '[]',
94 freshness_json TEXT,
95 row_hash TEXT,
96 source_watermark TEXT,
97 PRIMARY KEY (from_id, to_id, kind),
98 FOREIGN KEY (from_id) REFERENCES graph_nodes(id) ON DELETE CASCADE,
99 FOREIGN KEY (to_id) REFERENCES graph_nodes(id) ON DELETE CASCADE
100 );
101 CREATE INDEX IF NOT EXISTS idx_graph_edges_from_kind
102 ON graph_edges(from_id, kind);
103 CREATE INDEX IF NOT EXISTS idx_graph_edges_to_kind
104 ON graph_edges(to_id, kind);
105
106 CREATE TABLE IF NOT EXISTS graph_node_properties (
107 node_id TEXT NOT NULL,
108 key TEXT NOT NULL,
109 value TEXT NOT NULL,
110 PRIMARY KEY (node_id, key),
111 FOREIGN KEY (node_id) REFERENCES graph_nodes(id) ON DELETE CASCADE
112 );
113 CREATE INDEX IF NOT EXISTS idx_graph_node_properties_key_value_node
114 ON graph_node_properties(key, value, node_id);
115
116 CREATE TABLE IF NOT EXISTS graph_edge_properties (
117 edge_key TEXT NOT NULL,
118 key TEXT NOT NULL,
119 value TEXT NOT NULL,
120 PRIMARY KEY (edge_key, key),
121 FOREIGN KEY (edge_key) REFERENCES graph_edges(edge_key) ON DELETE CASCADE
122 );
123 CREATE INDEX IF NOT EXISTS idx_graph_edge_properties_key_value_edge
124 ON graph_edge_properties(key, value, edge_key);
125
126 CREATE TABLE IF NOT EXISTS graph_projection_versions (
127 scope TEXT PRIMARY KEY,
128 projection_version TEXT NOT NULL,
129 content_hash TEXT,
130 source_watermark TEXT,
131 observed_at_unix INTEGER NOT NULL
132 );
133
134 CREATE TABLE IF NOT EXISTS graph_tombstones (
135 row_key TEXT PRIMARY KEY,
136 row_kind TEXT NOT NULL,
137 deleted_at_unix INTEGER NOT NULL
138 );
139
140 PRAGMA user_version = {SQLITE_GRAPH_SCHEMA_VERSION};
141 "#,
142 ))
143 .await
144 .context("initializing libsql graph schema")?;
145 Ok::<(), anyhow::Error>(())
146 })?;
147 Ok(())
148 }
149}
150
151fn to_json<T: serde::Serialize>(value: &T) -> Result<String> {
152 serde_json::to_string(value).map_err(Into::into)
153}
154
155fn optional_to_json<T: serde::Serialize>(value: &Option<T>) -> Result<Option<String>> {
156 value.as_ref().map(to_json).transpose()
157}
158
159fn node_from_row(row: &libsql::Row) -> Result<GraphNode> {
160 let id: String = row.get(0)?;
161 let kind: String = row.get(1)?;
162 let label: String = row.get(2)?;
163 let properties_json: String = row.get(3)?;
164 let provenance_json: String = row.get(4)?;
165 let freshness_json: Option<String> = row.get(5)?;
166 Ok(GraphNode {
167 id,
168 kind,
169 label,
170 properties: serde_json::from_str(&properties_json)?,
171 provenance: serde_json::from_str(&provenance_json)?,
172 freshness: freshness_json
173 .map(|v| serde_json::from_str(&v))
174 .transpose()?,
175 })
176}
177
178fn edge_from_row(row: &libsql::Row) -> Result<GraphEdge> {
179 let edge_key: String = row.get(0)?;
180 let from_id: String = row.get(1)?;
181 let to_id: String = row.get(2)?;
182 let kind: String = row.get(3)?;
183 let properties_json: String = row.get(4)?;
184 let provenance_json: String = row.get(5)?;
185 let freshness_json: Option<String> = row.get(6)?;
186 Ok(GraphEdge {
187 id: edge_key,
188 from_id,
189 to_id,
190 kind,
191 properties: serde_json::from_str(&properties_json)?,
192 provenance: serde_json::from_str(&provenance_json)?,
193 freshness: freshness_json
194 .map(|v| serde_json::from_str(&v))
195 .transpose()?,
196 })
197}
198
199fn stable_graph_edge_id(from_id: &str, to_id: &str, kind: &str) -> String {
200 let raw = serde_json::json!([from_id, kind, to_id]).to_string();
201 format!("edge:{}", blake3::hash(raw.as_bytes()).to_hex())
202}
203
204fn row_hash<T: serde::Serialize>(value: &T) -> Result<String> {
205 let payload = serde_json::to_vec(value)?;
206 Ok(blake3::hash(&payload).to_hex().to_string())
207}
208
209fn replace_node_properties(
210 conn: &libsql::Connection,
211 rt: &tokio::runtime::Runtime,
212 node_id: &str,
213 properties: &BTreeMap<String, String>,
214) -> Result<()> {
215 block_on(rt, async {
216 conn.execute(
217 "DELETE FROM graph_node_properties WHERE node_id = ?1",
218 [node_id],
219 )
220 .await?;
221 let stmt = conn
222 .prepare("INSERT INTO graph_node_properties (node_id, key, value) VALUES (?1, ?2, ?3)")
223 .await?;
224 for (key, value) in properties {
225 stmt.execute(libsql::params![
226 node_id.to_string(),
227 key.clone(),
228 value.clone()
229 ])
230 .await?;
231 }
232 Ok::<(), anyhow::Error>(())
233 })?;
234 Ok(())
235}
236
237fn replace_edge_properties(
238 conn: &libsql::Connection,
239 rt: &tokio::runtime::Runtime,
240 edge_key: &str,
241 properties: &BTreeMap<String, String>,
242) -> Result<()> {
243 block_on(rt, async {
244 conn.execute(
245 "DELETE FROM graph_edge_properties WHERE edge_key = ?1",
246 [edge_key.to_string()],
247 )
248 .await?;
249 let stmt = conn
250 .prepare("INSERT INTO graph_edge_properties (edge_key, key, value) VALUES (?1, ?2, ?3)")
251 .await?;
252 for (key, value) in properties {
253 stmt.execute(libsql::params![
254 edge_key.to_string(),
255 key.clone(),
256 value.clone()
257 ])
258 .await?;
259 }
260 Ok::<(), anyhow::Error>(())
261 })?;
262 Ok(())
263}
264
265impl GraphStore for LibsqlGraphStore {
266 fn upsert_node(&self, node: &GraphNode) -> Result<()> {
267 let id = node.id.clone();
268 block_on(&self.rt, async {
269 let properties_json = to_json(&node.properties)?;
270 let provenance_json = to_json(&node.provenance)?;
271 let freshness_json = optional_to_json(&node.freshness)?;
272 let hash = row_hash(node)?;
273 self.conn.execute(
274 r#"
275 INSERT INTO graph_nodes
276 (id, kind, label, properties_json, provenance_json, freshness_json, row_hash, source_watermark)
277 VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, NULL)
278 ON CONFLICT(id) DO UPDATE SET
279 kind = excluded.kind,
280 label = excluded.label,
281 properties_json = excluded.properties_json,
282 provenance_json = excluded.provenance_json,
283 freshness_json = excluded.freshness_json,
284 row_hash = excluded.row_hash,
285 source_watermark = excluded.source_watermark
286 "#,
287 libsql::params![node.id.clone(), node.kind.clone(), node.label.clone(), properties_json, provenance_json, freshness_json, hash],
288 ).await?;
289 Ok::<(), anyhow::Error>(())
290 })?;
291 replace_node_properties(&self.conn, &self.rt, &id, &node.properties)?;
292 Ok(())
293 }
294
295 fn upsert_edge(&self, edge: &GraphEdge) -> Result<()> {
296 let edge_key = if edge.id.is_empty() {
297 stable_graph_edge_id(&edge.from_id, &edge.to_id, &edge.kind)
298 } else {
299 edge.id.clone()
300 };
301 let edge_key_for_props = edge_key.clone();
302 block_on(&self.rt, async {
303 let properties_json = to_json(&edge.properties)?;
304 let provenance_json = to_json(&edge.provenance)?;
305 let freshness_json = optional_to_json(&edge.freshness)?;
306 let hash = row_hash(edge)?;
307 self.conn.execute(
308 r#"
309 INSERT INTO graph_edges
310 (edge_key, from_id, to_id, kind, properties_json, provenance_json, freshness_json, row_hash, source_watermark)
311 VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, NULL)
312 ON CONFLICT(from_id, to_id, kind) DO UPDATE SET
313 edge_key = excluded.edge_key,
314 properties_json = excluded.properties_json,
315 provenance_json = excluded.provenance_json,
316 freshness_json = excluded.freshness_json,
317 row_hash = excluded.row_hash,
318 source_watermark = excluded.source_watermark
319 "#,
320 libsql::params![edge_key, edge.from_id.clone(), edge.to_id.clone(), edge.kind.clone(), properties_json, provenance_json, freshness_json, hash],
321 ).await?;
322 Ok::<(), anyhow::Error>(())
323 })?;
324 replace_edge_properties(&self.conn, &self.rt, &edge_key_for_props, &edge.properties)?;
325 Ok(())
326 }
327
328 fn delete_node(&self, id: &str) -> Result<usize> {
329 let count = block_on(&self.rt, async {
330 let result = self
331 .conn
332 .execute("DELETE FROM graph_nodes WHERE id = ?1", [id])
333 .await?;
334 Ok::<u64, anyhow::Error>(result)
335 })?;
336 Ok(count as usize)
337 }
338
339 fn delete_edge(&self, from_id: &str, to_id: &str, kind: &str) -> Result<usize> {
340 let count = block_on(&self.rt, async {
341 let result = self
342 .conn
343 .execute(
344 "DELETE FROM graph_edges WHERE from_id = ?1 AND to_id = ?2 AND kind = ?3",
345 libsql::params![from_id, to_id, kind],
346 )
347 .await?;
348 Ok::<u64, anyhow::Error>(result)
349 })?;
350 Ok(count as usize)
351 }
352
353 fn node(&self, id: &str) -> Result<Option<GraphNode>> {
354 block_on(&self.rt, async {
355 let mut rows = self
356 .conn
357 .query(
358 r#"
359 SELECT id, kind, label, properties_json, provenance_json, freshness_json
360 FROM graph_nodes
361 WHERE id = ?1
362 "#,
363 [id],
364 )
365 .await?;
366 match rows.next().await? {
367 Some(row) => Ok(Some(node_from_row(&row)?)),
368 None => Ok(None),
369 }
370 })
371 }
372
373 fn all_nodes(&self) -> Result<Vec<GraphNode>> {
374 block_on(&self.rt, async {
375 let mut rows = self
376 .conn
377 .query(
378 r#"
379 SELECT id, kind, label, properties_json, provenance_json, freshness_json
380 FROM graph_nodes
381 ORDER BY id
382 "#,
383 (),
384 )
385 .await?;
386 let mut nodes = Vec::new();
387 while let Some(row) = rows.next().await? {
388 nodes.push(node_from_row(&row)?);
389 }
390 Ok(nodes)
391 })
392 }
393
394 fn all_edges(&self) -> Result<Vec<GraphEdge>> {
395 block_on(&self.rt, async {
396 let mut rows = self.conn.query(
397 r#"
398 SELECT edge_key, from_id, to_id, kind, properties_json, provenance_json, freshness_json
399 FROM graph_edges
400 ORDER BY from_id, kind, to_id
401 "#,
402 (),
403 ).await?;
404 let mut edges = Vec::new();
405 while let Some(row) = rows.next().await? {
406 edges.push(edge_from_row(&row)?);
407 }
408 Ok(edges)
409 })
410 }
411
412 fn edge(&self, edge_id: &str) -> Result<Option<GraphEdge>> {
413 block_on(&self.rt, async {
414 let mut rows = self
415 .conn
416 .query(
417 r#"
418 SELECT edge_key, from_id, to_id, kind, properties_json, provenance_json, freshness_json
419 FROM graph_edges
420 WHERE edge_key = ?1
421 "#,
422 [edge_id],
423 )
424 .await?;
425 match rows.next().await? {
426 Some(row) => Ok(Some(edge_from_row(&row)?)),
427 None => Ok(None),
428 }
429 })
430 }
431
432 fn graph_counts(&self) -> Result<(usize, usize)> {
433 let nodes = block_on(&self.rt, async {
434 let mut rows = self
435 .conn
436 .query("SELECT COUNT(*) FROM graph_nodes", ())
437 .await?;
438 let row = rows.next().await?.context("no row")?;
439 let count: u64 = row.get(0)?;
440 Ok::<u64, anyhow::Error>(count)
441 })?;
442 let edges = block_on(&self.rt, async {
443 let mut rows = self
444 .conn
445 .query("SELECT COUNT(*) FROM graph_edges", ())
446 .await?;
447 let row = rows.next().await?.context("no row")?;
448 let count: u64 = row.get(0)?;
449 Ok::<u64, anyhow::Error>(count)
450 })?;
451 Ok((nodes as usize, edges as usize))
452 }
453
454 fn nodes_by_kind(&self, kind: &str) -> Result<Vec<GraphNode>> {
455 block_on(&self.rt, async {
456 let mut rows = self
457 .conn
458 .query(
459 r#"
460 SELECT id, kind, label, properties_json, provenance_json, freshness_json
461 FROM graph_nodes
462 WHERE kind = ?1
463 ORDER BY id
464 "#,
465 [kind],
466 )
467 .await?;
468 let mut nodes = Vec::new();
469 while let Some(row) = rows.next().await? {
470 nodes.push(node_from_row(&row)?);
471 }
472 Ok(nodes)
473 })
474 }
475
476 fn outgoing_edges(&self, from_id: &str, kind: Option<&str>) -> Result<Vec<GraphEdge>> {
477 block_on(&self.rt, async {
478 let mut edges = Vec::new();
479 match kind {
480 Some(kind) => {
481 let mut rows = self.conn.query(
482 r#"
483 SELECT edge_key, from_id, to_id, kind, properties_json, provenance_json, freshness_json
484 FROM graph_edges
485 WHERE from_id = ?1 AND kind = ?2
486 ORDER BY to_id, kind
487 "#,
488 libsql::params![from_id, kind],
489 ).await?;
490 while let Some(row) = rows.next().await? {
491 edges.push(edge_from_row(&row)?);
492 }
493 }
494 None => {
495 let mut rows = self.conn.query(
496 r#"
497 SELECT edge_key, from_id, to_id, kind, properties_json, provenance_json, freshness_json
498 FROM graph_edges
499 WHERE from_id = ?1
500 ORDER BY to_id, kind
501 "#,
502 [from_id],
503 ).await?;
504 while let Some(row) = rows.next().await? {
505 edges.push(edge_from_row(&row)?);
506 }
507 }
508 }
509 Ok(edges)
510 })
511 }
512
513 fn incident_edges(&self, node_id: &str, kind: Option<&str>) -> Result<Vec<GraphEdge>> {
514 block_on(&self.rt, async {
515 let sql = match kind {
516 Some(_) => {
517 r#"
518 SELECT edge_key, from_id, to_id, kind, properties_json, provenance_json, freshness_json
519 FROM (
520 SELECT edge_key, from_id, to_id, kind, properties_json, provenance_json, freshness_json
521 FROM graph_edges
522 WHERE from_id = ?1 AND kind = ?2
523 UNION
524 SELECT edge_key, from_id, to_id, kind, properties_json, provenance_json, freshness_json
525 FROM graph_edges
526 WHERE to_id = ?1 AND kind = ?2
527 ) e
528 ORDER BY e.edge_key
529 "#
530 }
531 None => {
532 r#"
533 SELECT edge_key, from_id, to_id, kind, properties_json, provenance_json, freshness_json
534 FROM (
535 SELECT edge_key, from_id, to_id, kind, properties_json, provenance_json, freshness_json
536 FROM graph_edges
537 WHERE from_id = ?1
538 UNION
539 SELECT edge_key, from_id, to_id, kind, properties_json, provenance_json, freshness_json
540 FROM graph_edges
541 WHERE to_id = ?1
542 ) e
543 ORDER BY e.edge_key
544 "#
545 }
546 };
547 let mut edges = Vec::new();
548 match kind {
549 Some(kind) => {
550 let mut rows = self.conn.query(sql, libsql::params![node_id, kind]).await?;
551 while let Some(row) = rows.next().await? {
552 edges.push(edge_from_row(&row)?);
553 }
554 }
555 None => {
556 let mut rows = self.conn.query(sql, [node_id]).await?;
557 while let Some(row) = rows.next().await? {
558 edges.push(edge_from_row(&row)?);
559 }
560 }
561 }
562 Ok(edges)
563 })
564 }
565
566 fn edges_between_nodes(&self, node_ids: &BTreeSet<String>) -> Result<Vec<GraphEdge>> {
567 if node_ids.is_empty() {
568 return Ok(Vec::new());
569 }
570 block_on(&self.rt, async {
571 self.conn
572 .execute_batch(
573 r#"
574 CREATE TEMP TABLE IF NOT EXISTS _edges_between_ids (id TEXT PRIMARY KEY);
575 DELETE FROM _edges_between_ids;
576 "#,
577 )
578 .await?;
579 for chunk in node_ids.iter().collect::<Vec<_>>().chunks(450) {
580 let row_placeholders: Vec<String> =
581 chunk.iter().map(|_| "(?)".to_string()).collect();
582 let placeholders = row_placeholders.join(", ");
583 let sql =
584 format!("INSERT OR IGNORE INTO _edges_between_ids (id) VALUES {placeholders}");
585 let values: Vec<String> = chunk.iter().map(|id| (*id).clone()).collect();
586 self.conn
587 .execute(
588 &sql,
589 libsql::params_from_iter(values.iter().map(|v| v.as_str())),
590 )
591 .await?;
592 }
593 let mut rows = self
594 .conn
595 .query(
596 r#"
597 SELECT e.edge_key, e.from_id, e.to_id, e.kind, e.properties_json, e.provenance_json, e.freshness_json
598 FROM graph_edges e
599 WHERE EXISTS (SELECT 1 FROM _edges_between_ids f WHERE f.id = e.from_id)
600 AND EXISTS (SELECT 1 FROM _edges_between_ids t WHERE t.id = e.to_id)
601 ORDER BY e.from_id, e.kind, e.to_id
602 "#,
603 (),
604 )
605 .await?;
606 let mut edges = Vec::new();
607 while let Some(row) = rows.next().await? {
608 edges.push(edge_from_row(&row)?);
609 }
610 Ok::<Vec<GraphEdge>, anyhow::Error>(edges)
611 })
612 }
613
614 fn shortest_path(
615 &self,
616 from_id: &str,
617 to_id: &str,
618 kind: Option<&str>,
619 ) -> Result<Option<GraphPath>> {
620 self.shortest_path_with_max_hops(from_id, to_id, kind, None)
621 }
622
623 fn shortest_path_with_max_hops(
624 &self,
625 from_id: &str,
626 to_id: &str,
627 kind: Option<&str>,
628 max_hops: Option<usize>,
629 ) -> Result<Option<GraphPath>> {
630 if from_id == to_id {
631 return Ok(Some(GraphPath {
632 nodes: vec![from_id.to_string()],
633 hops: 0,
634 }));
635 }
636 let hop_limit = max_hops.unwrap_or(usize::MAX);
637 if hop_limit == 0 {
638 return Ok(None);
639 }
640
641 let mut visited = BTreeSet::from([from_id.to_string()]);
642 let mut parent = BTreeMap::<String, String>::from([(from_id.to_string(), String::new())]);
643 let mut frontier = vec![from_id.to_string()];
644
645 for _depth in 0..hop_limit {
646 if frontier.is_empty() {
647 break;
648 }
649 let mut next_frontier = BTreeSet::new();
650 for current in &frontier {
651 let neighbors = self.outgoing_edges(current, kind)?;
652 for edge in neighbors {
653 if !visited.insert(edge.to_id.clone()) {
654 continue;
655 }
656 parent.insert(edge.to_id.clone(), current.clone());
657 if edge.to_id == to_id {
658 let mut nodes = vec![to_id.to_string()];
659 let mut cursor = to_id;
660 while let Some(previous) = parent.get(cursor) {
661 if previous.is_empty() {
662 break;
663 }
664 nodes.push(previous.clone());
665 cursor = previous;
666 }
667 nodes.reverse();
668 return Ok(Some(GraphPath {
669 hops: nodes.len().saturating_sub(1),
670 nodes,
671 }));
672 }
673 next_frontier.insert(edge.to_id);
674 }
675 }
676 frontier = next_frontier.into_iter().collect();
677 }
678 Ok(None)
679 }
680}
681
682#[cfg(test)]
683mod tests {
684 use super::*;
685 use tsift_core::{GraphFreshness, GraphProjection, GraphProvenance};
686
687 fn sample_provenance() -> GraphProvenance {
688 GraphProvenance::new("fixture", "src/lib.rs:1").with_content_hash("hash-1")
689 }
690
691 fn sample_projection() -> GraphProjection {
692 let source = sample_provenance();
693 GraphProjection {
694 nodes: vec![
695 GraphNode::new("doc:livekit", "document", "LiveKit guide")
696 .with_property("domain", "livekit")
697 .with_provenance(source.clone())
698 .with_freshness(GraphFreshness::content_hash("node-hash")),
699 GraphNode::new("topic:rooms", "topic", "Rooms"),
700 GraphNode::new("topic:egress", "topic", "Egress"),
701 ],
702 edges: vec![
703 GraphEdge::new("doc:livekit", "topic:rooms", "mentions")
704 .with_property("confidence", "0.91")
705 .with_provenance(source.clone())
706 .with_freshness(GraphFreshness::content_hash("edge-hash")),
707 GraphEdge::new("topic:rooms", "topic:egress", "related_to").with_provenance(source),
708 ],
709 }
710 }
711
712 #[test]
713 fn libsql_store_round_trips_generic_nodes_edges() {
714 let store = LibsqlGraphStore::in_memory().unwrap();
715 let source = sample_provenance();
716 let node = GraphNode::new("doc:livekit", "document", "LiveKit guide")
717 .with_property("domain", "livekit")
718 .with_provenance(source.clone())
719 .with_freshness(GraphFreshness::content_hash("node-hash"));
720 let topic = GraphNode::new("topic:rooms", "topic", "Rooms");
721 let edge = GraphEdge::new("doc:livekit", "topic:rooms", "mentions")
722 .with_property("confidence", "0.91")
723 .with_provenance(source)
724 .with_freshness(GraphFreshness::content_hash("edge-hash"));
725
726 store.upsert_node(&node).unwrap();
727 store.upsert_node(&topic).unwrap();
728 store.upsert_edge(&edge).unwrap();
729
730 assert_eq!(store.node("doc:livekit").unwrap(), Some(node));
731 assert_eq!(store.nodes_by_kind("topic").unwrap(), vec![topic]);
732 assert_eq!(store.all_nodes().unwrap().len(), 2);
733 assert_eq!(store.all_edges().unwrap().len(), 1);
734 assert_eq!(
735 store
736 .outgoing_edges("doc:livekit", Some("mentions"))
737 .unwrap(),
738 vec![edge]
739 );
740 }
741
742 #[test]
743 fn libsql_store_supports_projection_upsert() {
744 let store = LibsqlGraphStore::in_memory().unwrap();
745 let projection = sample_projection();
746 projection.upsert_into(&store).unwrap();
747
748 assert_eq!(store.node("doc:livekit").unwrap().unwrap().kind, "document");
749 assert_eq!(store.nodes_by_kind("topic").unwrap().len(), 2);
750 let mentions = store
751 .outgoing_edges("doc:livekit", Some("mentions"))
752 .unwrap();
753 assert_eq!(mentions.len(), 1);
754 assert_eq!(mentions[0].to_id, "topic:rooms");
755 }
756
757 #[test]
758 fn libsql_store_crud_neighborhood_and_ordering() {
759 let store = LibsqlGraphStore::in_memory().unwrap();
760 let projection = sample_projection();
761 projection.upsert_into(&store).unwrap();
762
763 let neighborhood = store.neighborhood("doc:livekit", 2, None).unwrap().unwrap();
764 let node_ids: Vec<&str> = neighborhood.nodes.iter().map(|n| n.id.as_str()).collect();
765 assert_eq!(node_ids, vec!["doc:livekit", "topic:egress", "topic:rooms"]);
766
767 assert_eq!(
768 store
769 .delete_edge("topic:rooms", "topic:egress", "related_to")
770 .unwrap(),
771 1
772 );
773 assert!(
774 store
775 .shortest_path("doc:livekit", "topic:egress", None)
776 .unwrap()
777 .is_none()
778 );
779 assert_eq!(store.delete_node("topic:rooms").unwrap(), 1);
780 assert!(store.node("topic:rooms").unwrap().is_none());
781 assert!(
782 store
783 .outgoing_edges("doc:livekit", None)
784 .unwrap()
785 .is_empty()
786 );
787 }
788
789 #[test]
790 fn libsql_store_shortest_path() {
791 let store = LibsqlGraphStore::in_memory().unwrap();
792 for id in ["a", "b", "c"] {
793 store
794 .upsert_node(&GraphNode::new(id, "symbol", id))
795 .unwrap();
796 }
797 store
798 .upsert_edge(&GraphEdge::new("a", "b", "calls"))
799 .unwrap();
800 store
801 .upsert_edge(&GraphEdge::new("a", "c", "documents"))
802 .unwrap();
803 store
804 .upsert_edge(&GraphEdge::new("b", "c", "calls"))
805 .unwrap();
806
807 let calls = store.outgoing_edges("a", Some("calls")).unwrap();
808 assert_eq!(calls.len(), 1);
809 assert_eq!(calls[0].to_id, "b");
810
811 let path = store
812 .shortest_path("a", "c", Some("calls"))
813 .unwrap()
814 .unwrap();
815 assert_eq!(path.nodes, vec!["a", "b", "c"]);
816 assert_eq!(path.hops, 2);
817
818 assert!(
819 store
820 .shortest_path("c", "a", Some("calls"))
821 .unwrap()
822 .is_none()
823 );
824 }
825
826 #[test]
827 fn libsql_store_open_creates_db_file() {
828 let dir = tempfile::tempdir().unwrap();
829 let db_path = dir.path().join("test-graph.db");
830 let store = LibsqlGraphStore::open(&db_path).unwrap();
831 store
832 .upsert_node(&GraphNode::new("test", "test", "test"))
833 .unwrap();
834 assert!(db_path.exists());
835 }
836
837 #[test]
838 fn libsql_graph_counts() {
839 let store = LibsqlGraphStore::in_memory().unwrap();
840 for id in ["a", "b", "c"] {
841 store
842 .upsert_node(&GraphNode::new(id, "symbol", id))
843 .unwrap();
844 }
845 store
846 .upsert_edge(&GraphEdge::new("a", "b", "calls"))
847 .unwrap();
848 store
849 .upsert_edge(&GraphEdge::new("b", "c", "calls"))
850 .unwrap();
851 let (nodes, edges) = store.graph_counts().unwrap();
852 assert_eq!(nodes, 3);
853 assert_eq!(edges, 2);
854 }
855
856 #[test]
857 fn libsql_incident_edges_pushdown() {
858 let store = LibsqlGraphStore::in_memory().unwrap();
859 for id in ["a", "b", "c", "d"] {
860 store
861 .upsert_node(&GraphNode::new(id, "symbol", id))
862 .unwrap();
863 }
864 store
865 .upsert_edge(&GraphEdge::new("a", "b", "calls"))
866 .unwrap();
867 store
868 .upsert_edge(&GraphEdge::new("a", "c", "documents"))
869 .unwrap();
870 store
871 .upsert_edge(&GraphEdge::new("d", "b", "calls"))
872 .unwrap();
873 store
874 .upsert_edge(&GraphEdge::new("c", "b", "references"))
875 .unwrap();
876
877 let all_incident = store.incident_edges("b", None).unwrap();
878 assert_eq!(all_incident.len(), 3);
879
880 let calls_incident = store.incident_edges("b", Some("calls")).unwrap();
881 assert_eq!(calls_incident.len(), 2);
882 assert!(calls_incident.iter().all(|e| e.kind == "calls"));
883
884 let docs_incident = store.incident_edges("b", Some("documents")).unwrap();
885 assert!(docs_incident.is_empty());
886
887 let a_incident = store.incident_edges("a", None).unwrap();
888 assert_eq!(a_incident.len(), 2);
889 assert!(a_incident.iter().all(|e| e.from_id == "a"));
890
891 let d_incident = store.incident_edges("d", None).unwrap();
892 assert_eq!(d_incident.len(), 1);
893 assert_eq!(d_incident[0].to_id, "b");
894 }
895
896 #[test]
897 fn libsql_store_edges_between_nodes_pushdown() {
898 let store = LibsqlGraphStore::in_memory().unwrap();
899 for id in ["a", "b", "c", "outside"] {
900 store
901 .upsert_node(&GraphNode::new(id, "symbol", id))
902 .unwrap();
903 }
904 for edge in [
905 GraphEdge::new("a", "b", "calls"),
906 GraphEdge::new("b", "c", "calls"),
907 GraphEdge::new("a", "outside", "calls"),
908 GraphEdge::new("outside", "c", "calls"),
909 ] {
910 store.upsert_edge(&edge).unwrap();
911 }
912
913 let scoped = ["a".to_string(), "b".to_string(), "c".to_string()]
914 .into_iter()
915 .collect::<BTreeSet<_>>();
916 let edge_keys = store
917 .edges_between_nodes(&scoped)
918 .unwrap()
919 .into_iter()
920 .map(|edge| (edge.from_id, edge.kind, edge.to_id))
921 .collect::<Vec<_>>();
922
923 assert_eq!(
924 edge_keys,
925 vec![
926 ("a".to_string(), "calls".to_string(), "b".to_string()),
927 ("b".to_string(), "calls".to_string(), "c".to_string()),
928 ]
929 );
930
931 let empty: BTreeSet<String> = BTreeSet::new();
932 assert!(store.edges_between_nodes(&empty).unwrap().is_empty());
933
934 let single = ["a".to_string()].into_iter().collect::<BTreeSet<_>>();
935 assert!(store.edges_between_nodes(&single).unwrap().is_empty());
936 }
937}