use std::path::Path;
use std::sync::Mutex;
use rusqlite::Connection;
use crate::error::Result;
use crate::graph::recall::smart_recall;
use crate::graph::schema::{init_graph_schema, migrate_graph, schema_version};
use crate::graph::search::{query_nodes, search_nodes};
use crate::graph::store::{
append_edge, delete_edge, delete_node, edges_for_node, remove_edges_for_node, upsert_node,
};
use crate::graph::traversal::related_nodes;
use crate::graph::types::{EdgeDirection, GraphEdge, GraphNode, ScoredNode};
pub trait GraphBackend: Send + Sync {
fn upsert_node(&self, node: &GraphNode) -> Result<()>;
fn read_node(&self, id: &str) -> Result<Option<GraphNode>>;
fn delete_node(&self, id: &str) -> Result<bool>;
fn search_nodes(&self, query: &str, limit: usize) -> Result<Vec<GraphNode>>;
#[allow(clippy::too_many_arguments)]
fn query_nodes(
&self,
tag: Option<&str>,
node_type: Option<&str>,
project: Option<&str>,
limit: usize,
) -> Result<Vec<GraphNode>>;
fn smart_recall(
&self,
project: Option<&str>,
hint: Option<&str>,
limit: usize,
) -> Result<Vec<ScoredNode>>;
fn related_nodes(&self, start_id: &str, depth: usize) -> Result<Vec<String>>;
fn append_edge(&self, edge: &GraphEdge) -> Result<()>;
fn append_edges(&self, edges: &[GraphEdge]) -> Result<()> {
for edge in edges {
self.append_edge(edge)?;
}
Ok(())
}
fn edges_for_node_dir(
&self,
node_id: &str,
dir: EdgeDirection,
relation: Option<&str>,
) -> Result<Vec<GraphEdge>> {
Ok(self
.edges_for_node(node_id)?
.into_iter()
.filter(|e| match dir {
EdgeDirection::Out => e.source == node_id,
EdgeDirection::In => e.target == node_id,
EdgeDirection::Both => true,
})
.filter(|e| relation.is_none_or(|r| e.relation == r))
.collect())
}
fn neighbors_weighted(
&self,
seed_ids: &[String],
dir: EdgeDirection,
relation: Option<&str>,
) -> Result<Vec<(String, f64)>> {
let mut weights: std::collections::HashMap<String, f64> = std::collections::HashMap::new();
let seed_set: std::collections::HashSet<&str> =
seed_ids.iter().map(String::as_str).collect();
for seed in seed_ids {
for e in self.edges_for_node_dir(seed, dir, relation)? {
let other = if e.source == *seed {
&e.target
} else {
&e.source
};
if !seed_set.contains(other.as_str()) {
*weights.entry(other.clone()).or_default() += e.weight;
}
}
}
let mut result: Vec<(String, f64)> = weights.into_iter().collect();
result.sort_by(|a, b| b.1.partial_cmp(&a.1).unwrap_or(std::cmp::Ordering::Equal));
Ok(result)
}
fn related_nodes_filtered(
&self,
start_id: &str,
depth: usize,
dir: EdgeDirection,
relation: Option<&str>,
) -> Result<Vec<String>> {
if depth == 0 {
return Ok(vec![]);
}
let mut visited: std::collections::HashSet<String> = std::collections::HashSet::new();
visited.insert(start_id.to_string());
let mut frontier: Vec<String> = vec![start_id.to_string()];
for _ in 0..depth {
let mut next: Vec<String> = Vec::new();
for node in &frontier {
for (nb, _) in self.neighbors_weighted(std::slice::from_ref(node), dir, relation)? {
if visited.insert(nb.clone()) {
next.push(nb);
}
}
}
if next.is_empty() {
break;
}
frontier = next;
}
visited.remove(start_id);
Ok(visited.into_iter().collect())
}
fn edges_for_node(&self, node_id: &str) -> Result<Vec<GraphEdge>>;
fn delete_edge(&self, id: &str) -> Result<bool>;
fn remove_edges_for_node(&self, node_id: &str) -> Result<()>;
fn current_version(&self) -> Result<u32>;
fn migrate(&self) -> Result<u32>;
}
pub struct SqliteGraph {
conn: Mutex<Connection>,
}
impl SqliteGraph {
pub fn open(path: impl AsRef<Path>) -> Result<Self> {
let conn = open_with_schema(path.as_ref())?;
Ok(Self {
conn: Mutex::new(conn),
})
}
pub fn open_in_memory() -> Result<Self> {
let conn = Connection::open_in_memory().map_err(store_err)?;
init_graph_schema(&conn)?;
let current = schema_version(&conn)?;
migrate_graph(&conn, current)?;
Ok(Self {
conn: Mutex::new(conn),
})
}
fn lock(&self) -> std::sync::MutexGuard<'_, Connection> {
self.conn.lock().unwrap_or_else(|e| e.into_inner())
}
}
fn open_with_schema(path: &Path) -> Result<Connection> {
let conn = Connection::open(path).map_err(store_err)?;
init_graph_schema(&conn)?;
let current = schema_version(&conn)?;
migrate_graph(&conn, current)?;
Ok(conn)
}
fn store_err(e: rusqlite::Error) -> crate::error::KernelError {
crate::error::KernelError::Store(e.to_string())
}
impl GraphBackend for SqliteGraph {
fn upsert_node(&self, node: &GraphNode) -> Result<()> {
let c = self.lock();
upsert_node(&c, node)
}
fn read_node(&self, id: &str) -> Result<Option<GraphNode>> {
let c = self.lock();
crate::graph::store::read_node(&c, id)
}
fn delete_node(&self, id: &str) -> Result<bool> {
let c = self.lock();
delete_node(&c, id)
}
fn search_nodes(&self, query: &str, limit: usize) -> Result<Vec<GraphNode>> {
let c = self.lock();
search_nodes(&c, query, limit)
}
fn query_nodes(
&self,
tag: Option<&str>,
node_type: Option<&str>,
project: Option<&str>,
limit: usize,
) -> Result<Vec<GraphNode>> {
let c = self.lock();
query_nodes(&c, tag, node_type, project, limit)
}
fn smart_recall(
&self,
project: Option<&str>,
hint: Option<&str>,
limit: usize,
) -> Result<Vec<ScoredNode>> {
let c = self.lock();
smart_recall(&c, project, hint, limit)
}
fn related_nodes(&self, start_id: &str, depth: usize) -> Result<Vec<String>> {
let c = self.lock();
Ok(related_nodes(&c, start_id, depth))
}
fn append_edge(&self, edge: &GraphEdge) -> Result<()> {
let c = self.lock();
append_edge(&c, edge)
}
fn append_edges(&self, edges: &[GraphEdge]) -> Result<()> {
let c = self.lock();
crate::graph::store::append_edges(&c, edges)
}
fn edges_for_node_dir(
&self,
node_id: &str,
dir: EdgeDirection,
relation: Option<&str>,
) -> Result<Vec<GraphEdge>> {
let c = self.lock();
crate::graph::store::edges_for_node_dir(&c, node_id, dir, relation)
}
fn neighbors_weighted(
&self,
seed_ids: &[String],
dir: EdgeDirection,
relation: Option<&str>,
) -> Result<Vec<(String, f64)>> {
let c = self.lock();
Ok(crate::graph::traversal::neighbors_weighted(
&c, seed_ids, dir, relation,
))
}
fn edges_for_node(&self, node_id: &str) -> Result<Vec<GraphEdge>> {
let c = self.lock();
edges_for_node(&c, node_id)
}
fn delete_edge(&self, id: &str) -> Result<bool> {
let c = self.lock();
delete_edge(&c, id)
}
fn remove_edges_for_node(&self, node_id: &str) -> Result<()> {
let c = self.lock();
remove_edges_for_node(&c, node_id)
}
fn current_version(&self) -> Result<u32> {
let c = self.lock();
schema_version(&c)
}
fn migrate(&self) -> Result<u32> {
let c = self.lock();
let current = schema_version(&c)?;
migrate_graph(&c, current)
}
}
#[cfg(feature = "graph-cjk")]
impl SqliteGraph {
pub fn search_nodes_cjk(&self, query: &str, limit: usize) -> Result<Vec<GraphNode>> {
let c = self.lock();
crate::graph::cjk::search_nodes_cjk(&c, query, limit)
}
pub fn segment_cjk(text: &str) -> String {
crate::graph::cjk::segment_cjk(text)
}
}
#[cfg(test)]
mod tests {
use super::*;
fn sample_node(id: &str) -> GraphNode {
GraphNode {
id: id.to_string(),
node_type: "concept".to_string(),
title: format!("Node {id}"),
body: "graph backend test body".to_string(),
tags: vec!["backend".to_string()],
projects: vec![],
agents: vec![],
created: "2026-01-01T00:00:00Z".to_string(),
updated: "2026-01-01T00:00:00Z".to_string(),
importance: 0.5,
access_count: 0,
accessed_at: String::new(),
}
}
#[test]
fn dyn_backend_round_trips_node() {
let backend: Box<dyn GraphBackend> = Box::new(SqliteGraph::open_in_memory().unwrap());
assert!(backend.read_node("n1").unwrap().is_none());
backend.upsert_node(&sample_node("n1")).unwrap();
let loaded = backend.read_node("n1").unwrap().unwrap();
assert_eq!(loaded.title, "Node n1");
assert_eq!(loaded.tags, vec!["backend".to_string()]);
assert!(backend.delete_node("n1").unwrap());
assert!(backend.read_node("n1").unwrap().is_none());
}
#[test]
fn fresh_backend_reports_current_version() {
let backend = SqliteGraph::open_in_memory().unwrap();
assert_eq!(
backend.current_version().unwrap(),
crate::graph::schema::GRAPH_SCHEMA_VERSION
);
}
#[test]
fn backend_search_finds_node() {
let backend = SqliteGraph::open_in_memory().unwrap();
backend.upsert_node(&sample_node("rust")).unwrap();
let hits = backend.search_nodes("graph backend", 10).unwrap();
assert_eq!(hits.len(), 1);
assert_eq!(hits[0].id, "rust");
}
#[test]
fn backend_smart_recall_finds_relevant() {
let backend = SqliteGraph::open_in_memory().unwrap();
let mut n = sample_node("rust");
n.body = "rust ownership borrow checker".to_string();
backend.upsert_node(&n).unwrap();
let recalled = backend.smart_recall(None, Some("ownership"), 5).unwrap();
assert!(recalled.iter().any(|s| s.node.id == "rust"));
}
#[test]
fn backend_related_nodes_traverses_edges() {
let backend = SqliteGraph::open_in_memory().unwrap();
backend.upsert_node(&sample_node("a")).unwrap();
backend.upsert_node(&sample_node("b")).unwrap();
backend
.append_edge(&GraphEdge {
id: "e1".into(),
source: "a".into(),
target: "b".into(),
relation: "related".into(),
weight: 1.0,
ts: "2026-01-01T00:00:00Z".into(),
})
.unwrap();
let related = backend.related_nodes("a", 2).unwrap();
assert!(related.contains(&"b".to_string()));
}
#[test]
fn dyn_backend_batch_and_filtered_edges() {
let backend: Box<dyn GraphBackend> = Box::new(SqliteGraph::open_in_memory().unwrap());
backend
.append_edges(&[
GraphEdge {
id: "e1".into(),
source: "a".into(),
target: "b".into(),
relation: "cites".into(),
weight: 1.0,
ts: "t".into(),
},
GraphEdge {
id: "e2".into(),
source: "c".into(),
target: "a".into(),
relation: "cites".into(),
weight: 1.0,
ts: "t".into(),
},
GraphEdge {
id: "e3".into(),
source: "a".into(),
target: "d".into(),
relation: "see_also".into(),
weight: 1.0,
ts: "t".into(),
},
])
.unwrap();
assert_eq!(
backend
.edges_for_node_dir("a", EdgeDirection::Out, None)
.unwrap()
.len(),
2
);
let nbs = backend
.neighbors_weighted(&["a".to_string()], EdgeDirection::Out, Some("cites"))
.unwrap();
let ids: Vec<&str> = nbs.iter().map(|(id, _)| id.as_str()).collect();
assert_eq!(ids, vec!["b"]);
let rel = backend
.related_nodes_filtered("a", 2, EdgeDirection::Out, None)
.unwrap();
assert!(rel.contains(&"b".to_string()));
assert!(rel.contains(&"d".to_string()));
}
}