use anyhow::{Context, Result};
use std::collections::{BTreeMap, BTreeSet};
use std::path::Path;
use tsift_core::{GraphEdge, GraphNode, GraphPath, GraphStore, SQLITE_GRAPH_SCHEMA_VERSION};
fn block_on<F: std::future::Future>(rt: &tokio::runtime::Runtime, f: F) -> F::Output {
rt.block_on(f)
}
pub struct LibsqlGraphStore {
conn: libsql::Connection,
rt: tokio::runtime::Runtime,
}
impl LibsqlGraphStore {
pub fn open(db_path: &Path) -> Result<Self> {
if let Some(parent) = db_path.parent() {
std::fs::create_dir_all(parent).with_context(|| {
format!("creating libsql graph substrate dir: {}", parent.display())
})?;
}
let rt = tokio::runtime::Runtime::new().context("creating tokio runtime for libsql")?;
let db = block_on(&rt, libsql::Builder::new_local(db_path).build())
.with_context(|| format!("opening libsql graph substrate db: {}", db_path.display()))?;
let conn = db.connect().with_context(|| {
format!(
"connecting to libsql graph substrate db: {}",
db_path.display()
)
})?;
let store = Self { conn, rt };
store.init_schema()?;
Ok(store)
}
pub fn open_remote(url: &str, auth_token: &str) -> Result<Self> {
let rt =
tokio::runtime::Runtime::new().context("creating tokio runtime for libsql remote")?;
let db = block_on(
&rt,
libsql::Builder::new_remote(url.to_string(), auth_token.to_string()).build(),
)
.context("building libsql remote database")?;
let conn = db
.connect()
.context("connecting to libsql remote database")?;
let store = Self { conn, rt };
store.init_schema()?;
Ok(store)
}
pub fn in_memory() -> Result<Self> {
let rt = tokio::runtime::Runtime::new().context("creating tokio runtime for libsql")?;
let db = block_on(&rt, libsql::Builder::new_local(":memory:").build())
.context("opening in-memory libsql database")?;
let conn = db
.connect()
.context("connecting to in-memory libsql database")?;
let store = Self { conn, rt };
store.init_schema()?;
Ok(store)
}
fn init_schema(&self) -> Result<()> {
block_on(&self.rt, async {
self.conn
.execute_batch(&format!(
r#"
PRAGMA foreign_keys = ON;
PRAGMA busy_timeout = 5000;
CREATE TABLE IF NOT EXISTS graph_nodes (
id TEXT PRIMARY KEY,
kind TEXT NOT NULL,
label TEXT NOT NULL,
properties_json TEXT NOT NULL DEFAULT '{{}}',
provenance_json TEXT NOT NULL DEFAULT '[]',
freshness_json TEXT,
row_hash TEXT,
source_watermark TEXT
);
CREATE INDEX IF NOT EXISTS idx_graph_nodes_kind
ON graph_nodes(kind);
CREATE INDEX IF NOT EXISTS idx_graph_nodes_kind_label
ON graph_nodes(kind, label, id);
CREATE TABLE IF NOT EXISTS graph_edges (
edge_key TEXT NOT NULL UNIQUE,
from_id TEXT NOT NULL,
to_id TEXT NOT NULL,
kind TEXT NOT NULL,
properties_json TEXT NOT NULL DEFAULT '{{}}',
provenance_json TEXT NOT NULL DEFAULT '[]',
freshness_json TEXT,
row_hash TEXT,
source_watermark TEXT,
PRIMARY KEY (from_id, to_id, kind),
FOREIGN KEY (from_id) REFERENCES graph_nodes(id) ON DELETE CASCADE,
FOREIGN KEY (to_id) REFERENCES graph_nodes(id) ON DELETE CASCADE
);
CREATE INDEX IF NOT EXISTS idx_graph_edges_from_kind
ON graph_edges(from_id, kind);
CREATE INDEX IF NOT EXISTS idx_graph_edges_to_kind
ON graph_edges(to_id, kind);
CREATE TABLE IF NOT EXISTS graph_node_properties (
node_id TEXT NOT NULL,
key TEXT NOT NULL,
value TEXT NOT NULL,
PRIMARY KEY (node_id, key),
FOREIGN KEY (node_id) REFERENCES graph_nodes(id) ON DELETE CASCADE
);
CREATE INDEX IF NOT EXISTS idx_graph_node_properties_key_value_node
ON graph_node_properties(key, value, node_id);
CREATE TABLE IF NOT EXISTS graph_edge_properties (
edge_key TEXT NOT NULL,
key TEXT NOT NULL,
value TEXT NOT NULL,
PRIMARY KEY (edge_key, key),
FOREIGN KEY (edge_key) REFERENCES graph_edges(edge_key) ON DELETE CASCADE
);
CREATE INDEX IF NOT EXISTS idx_graph_edge_properties_key_value_edge
ON graph_edge_properties(key, value, edge_key);
CREATE TABLE IF NOT EXISTS graph_projection_versions (
scope TEXT PRIMARY KEY,
projection_version TEXT NOT NULL,
content_hash TEXT,
source_watermark TEXT,
observed_at_unix INTEGER NOT NULL
);
CREATE TABLE IF NOT EXISTS graph_tombstones (
row_key TEXT PRIMARY KEY,
row_kind TEXT NOT NULL,
deleted_at_unix INTEGER NOT NULL
);
PRAGMA user_version = {SQLITE_GRAPH_SCHEMA_VERSION};
"#,
))
.await
.context("initializing libsql graph schema")?;
Ok::<(), anyhow::Error>(())
})?;
Ok(())
}
}
fn to_json<T: serde::Serialize>(value: &T) -> Result<String> {
serde_json::to_string(value).map_err(Into::into)
}
fn optional_to_json<T: serde::Serialize>(value: &Option<T>) -> Result<Option<String>> {
value.as_ref().map(to_json).transpose()
}
fn node_from_row(row: &libsql::Row) -> Result<GraphNode> {
let id: String = row.get(0)?;
let kind: String = row.get(1)?;
let label: String = row.get(2)?;
let properties_json: String = row.get(3)?;
let provenance_json: String = row.get(4)?;
let freshness_json: Option<String> = row.get(5)?;
Ok(GraphNode {
id,
kind,
label,
properties: serde_json::from_str(&properties_json)?,
provenance: serde_json::from_str(&provenance_json)?,
freshness: freshness_json
.map(|v| serde_json::from_str(&v))
.transpose()?,
})
}
fn edge_from_row(row: &libsql::Row) -> Result<GraphEdge> {
let edge_key: String = row.get(0)?;
let from_id: String = row.get(1)?;
let to_id: String = row.get(2)?;
let kind: String = row.get(3)?;
let properties_json: String = row.get(4)?;
let provenance_json: String = row.get(5)?;
let freshness_json: Option<String> = row.get(6)?;
Ok(GraphEdge {
id: edge_key,
from_id,
to_id,
kind,
properties: serde_json::from_str(&properties_json)?,
provenance: serde_json::from_str(&provenance_json)?,
freshness: freshness_json
.map(|v| serde_json::from_str(&v))
.transpose()?,
})
}
fn stable_graph_edge_id(from_id: &str, to_id: &str, kind: &str) -> String {
let raw = serde_json::json!([from_id, kind, to_id]).to_string();
format!("edge:{}", blake3::hash(raw.as_bytes()).to_hex())
}
fn row_hash<T: serde::Serialize>(value: &T) -> Result<String> {
let payload = serde_json::to_vec(value)?;
Ok(blake3::hash(&payload).to_hex().to_string())
}
fn replace_node_properties(
conn: &libsql::Connection,
rt: &tokio::runtime::Runtime,
node_id: &str,
properties: &BTreeMap<String, String>,
) -> Result<()> {
block_on(rt, async {
conn.execute(
"DELETE FROM graph_node_properties WHERE node_id = ?1",
[node_id],
)
.await?;
let stmt = conn
.prepare("INSERT INTO graph_node_properties (node_id, key, value) VALUES (?1, ?2, ?3)")
.await?;
for (key, value) in properties {
stmt.execute(libsql::params![
node_id.to_string(),
key.clone(),
value.clone()
])
.await?;
}
Ok::<(), anyhow::Error>(())
})?;
Ok(())
}
fn replace_edge_properties(
conn: &libsql::Connection,
rt: &tokio::runtime::Runtime,
edge_key: &str,
properties: &BTreeMap<String, String>,
) -> Result<()> {
block_on(rt, async {
conn.execute(
"DELETE FROM graph_edge_properties WHERE edge_key = ?1",
[edge_key.to_string()],
)
.await?;
let stmt = conn
.prepare("INSERT INTO graph_edge_properties (edge_key, key, value) VALUES (?1, ?2, ?3)")
.await?;
for (key, value) in properties {
stmt.execute(libsql::params![
edge_key.to_string(),
key.clone(),
value.clone()
])
.await?;
}
Ok::<(), anyhow::Error>(())
})?;
Ok(())
}
impl GraphStore for LibsqlGraphStore {
fn upsert_node(&self, node: &GraphNode) -> Result<()> {
let id = node.id.clone();
block_on(&self.rt, async {
let properties_json = to_json(&node.properties)?;
let provenance_json = to_json(&node.provenance)?;
let freshness_json = optional_to_json(&node.freshness)?;
let hash = row_hash(node)?;
self.conn.execute(
r#"
INSERT INTO graph_nodes
(id, kind, label, properties_json, provenance_json, freshness_json, row_hash, source_watermark)
VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, NULL)
ON CONFLICT(id) DO UPDATE SET
kind = excluded.kind,
label = excluded.label,
properties_json = excluded.properties_json,
provenance_json = excluded.provenance_json,
freshness_json = excluded.freshness_json,
row_hash = excluded.row_hash,
source_watermark = excluded.source_watermark
"#,
libsql::params![node.id.clone(), node.kind.clone(), node.label.clone(), properties_json, provenance_json, freshness_json, hash],
).await?;
Ok::<(), anyhow::Error>(())
})?;
replace_node_properties(&self.conn, &self.rt, &id, &node.properties)?;
Ok(())
}
fn upsert_edge(&self, edge: &GraphEdge) -> Result<()> {
let edge_key = if edge.id.is_empty() {
stable_graph_edge_id(&edge.from_id, &edge.to_id, &edge.kind)
} else {
edge.id.clone()
};
let edge_key_for_props = edge_key.clone();
block_on(&self.rt, async {
let properties_json = to_json(&edge.properties)?;
let provenance_json = to_json(&edge.provenance)?;
let freshness_json = optional_to_json(&edge.freshness)?;
let hash = row_hash(edge)?;
self.conn.execute(
r#"
INSERT INTO graph_edges
(edge_key, from_id, to_id, kind, properties_json, provenance_json, freshness_json, row_hash, source_watermark)
VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, NULL)
ON CONFLICT(from_id, to_id, kind) DO UPDATE SET
edge_key = excluded.edge_key,
properties_json = excluded.properties_json,
provenance_json = excluded.provenance_json,
freshness_json = excluded.freshness_json,
row_hash = excluded.row_hash,
source_watermark = excluded.source_watermark
"#,
libsql::params![edge_key, edge.from_id.clone(), edge.to_id.clone(), edge.kind.clone(), properties_json, provenance_json, freshness_json, hash],
).await?;
Ok::<(), anyhow::Error>(())
})?;
replace_edge_properties(&self.conn, &self.rt, &edge_key_for_props, &edge.properties)?;
Ok(())
}
fn delete_node(&self, id: &str) -> Result<usize> {
let count = block_on(&self.rt, async {
let result = self
.conn
.execute("DELETE FROM graph_nodes WHERE id = ?1", [id])
.await?;
Ok::<u64, anyhow::Error>(result)
})?;
Ok(count as usize)
}
fn delete_edge(&self, from_id: &str, to_id: &str, kind: &str) -> Result<usize> {
let count = block_on(&self.rt, async {
let result = self
.conn
.execute(
"DELETE FROM graph_edges WHERE from_id = ?1 AND to_id = ?2 AND kind = ?3",
libsql::params![from_id, to_id, kind],
)
.await?;
Ok::<u64, anyhow::Error>(result)
})?;
Ok(count as usize)
}
fn node(&self, id: &str) -> Result<Option<GraphNode>> {
block_on(&self.rt, async {
let mut rows = self
.conn
.query(
r#"
SELECT id, kind, label, properties_json, provenance_json, freshness_json
FROM graph_nodes
WHERE id = ?1
"#,
[id],
)
.await?;
match rows.next().await? {
Some(row) => Ok(Some(node_from_row(&row)?)),
None => Ok(None),
}
})
}
fn all_nodes(&self) -> Result<Vec<GraphNode>> {
block_on(&self.rt, async {
let mut rows = self
.conn
.query(
r#"
SELECT id, kind, label, properties_json, provenance_json, freshness_json
FROM graph_nodes
ORDER BY id
"#,
(),
)
.await?;
let mut nodes = Vec::new();
while let Some(row) = rows.next().await? {
nodes.push(node_from_row(&row)?);
}
Ok(nodes)
})
}
fn all_edges(&self) -> Result<Vec<GraphEdge>> {
block_on(&self.rt, async {
let mut rows = self.conn.query(
r#"
SELECT edge_key, from_id, to_id, kind, properties_json, provenance_json, freshness_json
FROM graph_edges
ORDER BY from_id, kind, to_id
"#,
(),
).await?;
let mut edges = Vec::new();
while let Some(row) = rows.next().await? {
edges.push(edge_from_row(&row)?);
}
Ok(edges)
})
}
fn edge(&self, edge_id: &str) -> Result<Option<GraphEdge>> {
block_on(&self.rt, async {
let mut rows = self
.conn
.query(
r#"
SELECT edge_key, from_id, to_id, kind, properties_json, provenance_json, freshness_json
FROM graph_edges
WHERE edge_key = ?1
"#,
[edge_id],
)
.await?;
match rows.next().await? {
Some(row) => Ok(Some(edge_from_row(&row)?)),
None => Ok(None),
}
})
}
fn graph_counts(&self) -> Result<(usize, usize)> {
let nodes = block_on(&self.rt, async {
let mut rows = self
.conn
.query("SELECT COUNT(*) FROM graph_nodes", ())
.await?;
let row = rows.next().await?.context("no row")?;
let count: u64 = row.get(0)?;
Ok::<u64, anyhow::Error>(count)
})?;
let edges = block_on(&self.rt, async {
let mut rows = self
.conn
.query("SELECT COUNT(*) FROM graph_edges", ())
.await?;
let row = rows.next().await?.context("no row")?;
let count: u64 = row.get(0)?;
Ok::<u64, anyhow::Error>(count)
})?;
Ok((nodes as usize, edges as usize))
}
fn nodes_by_kind(&self, kind: &str) -> Result<Vec<GraphNode>> {
block_on(&self.rt, async {
let mut rows = self
.conn
.query(
r#"
SELECT id, kind, label, properties_json, provenance_json, freshness_json
FROM graph_nodes
WHERE kind = ?1
ORDER BY id
"#,
[kind],
)
.await?;
let mut nodes = Vec::new();
while let Some(row) = rows.next().await? {
nodes.push(node_from_row(&row)?);
}
Ok(nodes)
})
}
fn outgoing_edges(&self, from_id: &str, kind: Option<&str>) -> Result<Vec<GraphEdge>> {
block_on(&self.rt, async {
let mut edges = Vec::new();
match kind {
Some(kind) => {
let mut rows = self.conn.query(
r#"
SELECT edge_key, from_id, to_id, kind, properties_json, provenance_json, freshness_json
FROM graph_edges
WHERE from_id = ?1 AND kind = ?2
ORDER BY to_id, kind
"#,
libsql::params![from_id, kind],
).await?;
while let Some(row) = rows.next().await? {
edges.push(edge_from_row(&row)?);
}
}
None => {
let mut rows = self.conn.query(
r#"
SELECT edge_key, from_id, to_id, kind, properties_json, provenance_json, freshness_json
FROM graph_edges
WHERE from_id = ?1
ORDER BY to_id, kind
"#,
[from_id],
).await?;
while let Some(row) = rows.next().await? {
edges.push(edge_from_row(&row)?);
}
}
}
Ok(edges)
})
}
fn incident_edges(&self, node_id: &str, kind: Option<&str>) -> Result<Vec<GraphEdge>> {
block_on(&self.rt, async {
let sql = match kind {
Some(_) => {
r#"
SELECT edge_key, from_id, to_id, kind, properties_json, provenance_json, freshness_json
FROM (
SELECT edge_key, from_id, to_id, kind, properties_json, provenance_json, freshness_json
FROM graph_edges
WHERE from_id = ?1 AND kind = ?2
UNION
SELECT edge_key, from_id, to_id, kind, properties_json, provenance_json, freshness_json
FROM graph_edges
WHERE to_id = ?1 AND kind = ?2
) e
ORDER BY e.edge_key
"#
}
None => {
r#"
SELECT edge_key, from_id, to_id, kind, properties_json, provenance_json, freshness_json
FROM (
SELECT edge_key, from_id, to_id, kind, properties_json, provenance_json, freshness_json
FROM graph_edges
WHERE from_id = ?1
UNION
SELECT edge_key, from_id, to_id, kind, properties_json, provenance_json, freshness_json
FROM graph_edges
WHERE to_id = ?1
) e
ORDER BY e.edge_key
"#
}
};
let mut edges = Vec::new();
match kind {
Some(kind) => {
let mut rows = self.conn.query(sql, libsql::params![node_id, kind]).await?;
while let Some(row) = rows.next().await? {
edges.push(edge_from_row(&row)?);
}
}
None => {
let mut rows = self.conn.query(sql, [node_id]).await?;
while let Some(row) = rows.next().await? {
edges.push(edge_from_row(&row)?);
}
}
}
Ok(edges)
})
}
fn edges_between_nodes(&self, node_ids: &BTreeSet<String>) -> Result<Vec<GraphEdge>> {
if node_ids.is_empty() {
return Ok(Vec::new());
}
block_on(&self.rt, async {
self.conn
.execute_batch(
r#"
CREATE TEMP TABLE IF NOT EXISTS _edges_between_ids (id TEXT PRIMARY KEY);
DELETE FROM _edges_between_ids;
"#,
)
.await?;
for chunk in node_ids.iter().collect::<Vec<_>>().chunks(450) {
let row_placeholders: Vec<String> =
chunk.iter().map(|_| "(?)".to_string()).collect();
let placeholders = row_placeholders.join(", ");
let sql =
format!("INSERT OR IGNORE INTO _edges_between_ids (id) VALUES {placeholders}");
let values: Vec<String> = chunk.iter().map(|id| (*id).clone()).collect();
self.conn
.execute(
&sql,
libsql::params_from_iter(values.iter().map(|v| v.as_str())),
)
.await?;
}
let mut rows = self
.conn
.query(
r#"
SELECT e.edge_key, e.from_id, e.to_id, e.kind, e.properties_json, e.provenance_json, e.freshness_json
FROM graph_edges e
WHERE EXISTS (SELECT 1 FROM _edges_between_ids f WHERE f.id = e.from_id)
AND EXISTS (SELECT 1 FROM _edges_between_ids t WHERE t.id = e.to_id)
ORDER BY e.from_id, e.kind, e.to_id
"#,
(),
)
.await?;
let mut edges = Vec::new();
while let Some(row) = rows.next().await? {
edges.push(edge_from_row(&row)?);
}
Ok::<Vec<GraphEdge>, anyhow::Error>(edges)
})
}
fn shortest_path(
&self,
from_id: &str,
to_id: &str,
kind: Option<&str>,
) -> Result<Option<GraphPath>> {
self.shortest_path_with_max_hops(from_id, to_id, kind, None)
}
fn shortest_path_with_max_hops(
&self,
from_id: &str,
to_id: &str,
kind: Option<&str>,
max_hops: Option<usize>,
) -> Result<Option<GraphPath>> {
if from_id == to_id {
return Ok(Some(GraphPath {
nodes: vec![from_id.to_string()],
hops: 0,
}));
}
let hop_limit = max_hops.unwrap_or(usize::MAX);
if hop_limit == 0 {
return Ok(None);
}
let mut visited = BTreeSet::from([from_id.to_string()]);
let mut parent = BTreeMap::<String, String>::from([(from_id.to_string(), String::new())]);
let mut frontier = vec![from_id.to_string()];
for _depth in 0..hop_limit {
if frontier.is_empty() {
break;
}
let mut next_frontier = BTreeSet::new();
for current in &frontier {
let neighbors = self.outgoing_edges(current, kind)?;
for edge in neighbors {
if !visited.insert(edge.to_id.clone()) {
continue;
}
parent.insert(edge.to_id.clone(), current.clone());
if edge.to_id == to_id {
let mut nodes = vec![to_id.to_string()];
let mut cursor = to_id;
while let Some(previous) = parent.get(cursor) {
if previous.is_empty() {
break;
}
nodes.push(previous.clone());
cursor = previous;
}
nodes.reverse();
return Ok(Some(GraphPath {
hops: nodes.len().saturating_sub(1),
nodes,
}));
}
next_frontier.insert(edge.to_id);
}
}
frontier = next_frontier.into_iter().collect();
}
Ok(None)
}
}
#[cfg(test)]
mod tests {
use super::*;
use tsift_core::{GraphFreshness, GraphProjection, GraphProvenance};
fn sample_provenance() -> GraphProvenance {
GraphProvenance::new("fixture", "src/lib.rs:1").with_content_hash("hash-1")
}
fn sample_projection() -> GraphProjection {
let source = sample_provenance();
GraphProjection {
nodes: vec![
GraphNode::new("doc:livekit", "document", "LiveKit guide")
.with_property("domain", "livekit")
.with_provenance(source.clone())
.with_freshness(GraphFreshness::content_hash("node-hash")),
GraphNode::new("topic:rooms", "topic", "Rooms"),
GraphNode::new("topic:egress", "topic", "Egress"),
],
edges: vec![
GraphEdge::new("doc:livekit", "topic:rooms", "mentions")
.with_property("confidence", "0.91")
.with_provenance(source.clone())
.with_freshness(GraphFreshness::content_hash("edge-hash")),
GraphEdge::new("topic:rooms", "topic:egress", "related_to").with_provenance(source),
],
}
}
#[test]
fn libsql_store_round_trips_generic_nodes_edges() {
let store = LibsqlGraphStore::in_memory().unwrap();
let source = sample_provenance();
let node = GraphNode::new("doc:livekit", "document", "LiveKit guide")
.with_property("domain", "livekit")
.with_provenance(source.clone())
.with_freshness(GraphFreshness::content_hash("node-hash"));
let topic = GraphNode::new("topic:rooms", "topic", "Rooms");
let edge = GraphEdge::new("doc:livekit", "topic:rooms", "mentions")
.with_property("confidence", "0.91")
.with_provenance(source)
.with_freshness(GraphFreshness::content_hash("edge-hash"));
store.upsert_node(&node).unwrap();
store.upsert_node(&topic).unwrap();
store.upsert_edge(&edge).unwrap();
assert_eq!(store.node("doc:livekit").unwrap(), Some(node));
assert_eq!(store.nodes_by_kind("topic").unwrap(), vec![topic]);
assert_eq!(store.all_nodes().unwrap().len(), 2);
assert_eq!(store.all_edges().unwrap().len(), 1);
assert_eq!(
store
.outgoing_edges("doc:livekit", Some("mentions"))
.unwrap(),
vec![edge]
);
}
#[test]
fn libsql_store_supports_projection_upsert() {
let store = LibsqlGraphStore::in_memory().unwrap();
let projection = sample_projection();
projection.upsert_into(&store).unwrap();
assert_eq!(store.node("doc:livekit").unwrap().unwrap().kind, "document");
assert_eq!(store.nodes_by_kind("topic").unwrap().len(), 2);
let mentions = store
.outgoing_edges("doc:livekit", Some("mentions"))
.unwrap();
assert_eq!(mentions.len(), 1);
assert_eq!(mentions[0].to_id, "topic:rooms");
}
#[test]
fn libsql_store_crud_neighborhood_and_ordering() {
let store = LibsqlGraphStore::in_memory().unwrap();
let projection = sample_projection();
projection.upsert_into(&store).unwrap();
let neighborhood = store.neighborhood("doc:livekit", 2, None).unwrap().unwrap();
let node_ids: Vec<&str> = neighborhood.nodes.iter().map(|n| n.id.as_str()).collect();
assert_eq!(node_ids, vec!["doc:livekit", "topic:egress", "topic:rooms"]);
assert_eq!(
store
.delete_edge("topic:rooms", "topic:egress", "related_to")
.unwrap(),
1
);
assert!(
store
.shortest_path("doc:livekit", "topic:egress", None)
.unwrap()
.is_none()
);
assert_eq!(store.delete_node("topic:rooms").unwrap(), 1);
assert!(store.node("topic:rooms").unwrap().is_none());
assert!(
store
.outgoing_edges("doc:livekit", None)
.unwrap()
.is_empty()
);
}
#[test]
fn libsql_store_shortest_path() {
let store = LibsqlGraphStore::in_memory().unwrap();
for id in ["a", "b", "c"] {
store
.upsert_node(&GraphNode::new(id, "symbol", id))
.unwrap();
}
store
.upsert_edge(&GraphEdge::new("a", "b", "calls"))
.unwrap();
store
.upsert_edge(&GraphEdge::new("a", "c", "documents"))
.unwrap();
store
.upsert_edge(&GraphEdge::new("b", "c", "calls"))
.unwrap();
let calls = store.outgoing_edges("a", Some("calls")).unwrap();
assert_eq!(calls.len(), 1);
assert_eq!(calls[0].to_id, "b");
let path = store
.shortest_path("a", "c", Some("calls"))
.unwrap()
.unwrap();
assert_eq!(path.nodes, vec!["a", "b", "c"]);
assert_eq!(path.hops, 2);
assert!(
store
.shortest_path("c", "a", Some("calls"))
.unwrap()
.is_none()
);
}
#[test]
fn libsql_store_open_creates_db_file() {
let dir = tempfile::tempdir().unwrap();
let db_path = dir.path().join("test-graph.db");
let store = LibsqlGraphStore::open(&db_path).unwrap();
store
.upsert_node(&GraphNode::new("test", "test", "test"))
.unwrap();
assert!(db_path.exists());
}
#[test]
fn libsql_graph_counts() {
let store = LibsqlGraphStore::in_memory().unwrap();
for id in ["a", "b", "c"] {
store
.upsert_node(&GraphNode::new(id, "symbol", id))
.unwrap();
}
store
.upsert_edge(&GraphEdge::new("a", "b", "calls"))
.unwrap();
store
.upsert_edge(&GraphEdge::new("b", "c", "calls"))
.unwrap();
let (nodes, edges) = store.graph_counts().unwrap();
assert_eq!(nodes, 3);
assert_eq!(edges, 2);
}
#[test]
fn libsql_incident_edges_pushdown() {
let store = LibsqlGraphStore::in_memory().unwrap();
for id in ["a", "b", "c", "d"] {
store
.upsert_node(&GraphNode::new(id, "symbol", id))
.unwrap();
}
store
.upsert_edge(&GraphEdge::new("a", "b", "calls"))
.unwrap();
store
.upsert_edge(&GraphEdge::new("a", "c", "documents"))
.unwrap();
store
.upsert_edge(&GraphEdge::new("d", "b", "calls"))
.unwrap();
store
.upsert_edge(&GraphEdge::new("c", "b", "references"))
.unwrap();
let all_incident = store.incident_edges("b", None).unwrap();
assert_eq!(all_incident.len(), 3);
let calls_incident = store.incident_edges("b", Some("calls")).unwrap();
assert_eq!(calls_incident.len(), 2);
assert!(calls_incident.iter().all(|e| e.kind == "calls"));
let docs_incident = store.incident_edges("b", Some("documents")).unwrap();
assert!(docs_incident.is_empty());
let a_incident = store.incident_edges("a", None).unwrap();
assert_eq!(a_incident.len(), 2);
assert!(a_incident.iter().all(|e| e.from_id == "a"));
let d_incident = store.incident_edges("d", None).unwrap();
assert_eq!(d_incident.len(), 1);
assert_eq!(d_incident[0].to_id, "b");
}
#[test]
fn libsql_store_edges_between_nodes_pushdown() {
let store = LibsqlGraphStore::in_memory().unwrap();
for id in ["a", "b", "c", "outside"] {
store
.upsert_node(&GraphNode::new(id, "symbol", id))
.unwrap();
}
for edge in [
GraphEdge::new("a", "b", "calls"),
GraphEdge::new("b", "c", "calls"),
GraphEdge::new("a", "outside", "calls"),
GraphEdge::new("outside", "c", "calls"),
] {
store.upsert_edge(&edge).unwrap();
}
let scoped = ["a".to_string(), "b".to_string(), "c".to_string()]
.into_iter()
.collect::<BTreeSet<_>>();
let edge_keys = store
.edges_between_nodes(&scoped)
.unwrap()
.into_iter()
.map(|edge| (edge.from_id, edge.kind, edge.to_id))
.collect::<Vec<_>>();
assert_eq!(
edge_keys,
vec![
("a".to_string(), "calls".to_string(), "b".to_string()),
("b".to_string(), "calls".to_string(), "c".to_string()),
]
);
let empty: BTreeSet<String> = BTreeSet::new();
assert!(store.edges_between_nodes(&empty).unwrap().is_empty());
let single = ["a".to_string()].into_iter().collect::<BTreeSet<_>>();
assert!(store.edges_between_nodes(&single).unwrap().is_empty());
}
}