crates/knut-graphdb/src/lib.rs
//! # knut-graphdb — a backend-agnostic graph-database seam
//!
//! **One trait, [`GraphDb`], is the whole client surface a consumer needs over a
//! graph endpoint:** list its graphs, open one for its catalogue, read a scene,
//! expand a node's neighbours, find a bounded shortest path, upsert rows.
//!
//! > **A `GraphDb` is a CLIENT. It takes an endpoint and talks to it. It never
//! > starts a process and it never knows containers exist.**
//!
//! Backends render the ops in their **native** language:
//!
//! - [`falkor::FalkorBackend`] (feature `falkordb`, on by default) — emits Cypher
//! and runs it through graphar-flight's `FalkorExecutor` (a live FalkorDB /
//! Valkey container), optionally over mutual TLS.
//! - [`cozo::CozoBackend`] (feature `cozo`) — an **embedded** CozoDB (the `mem`
//! engine: no container, no redis, no socket) that answers the same ops in
//! CozoScript/Datalog.
//!
//! ## Why a trait of ops, not raw Cypher
//!
//! Cozo speaks **Datalog**, not Cypher, so a raw-Cypher seam could never target it.
//! Every base method is an *intent* ("the neighbours of node 42 in graph `g`") and
//! each backend compiles that intent to its own query language.
//!
//! ## Cypher is OPTIONAL — and it is not on the base trait at all
//!
//! [`GraphDb::read`] takes a graph name and a row cap. **No query language appears
//! anywhere in the base trait's signatures**, so a backend with no query language (a
//! GraphAr file set, a knut projection, an in-memory scene) is a complete, valid
//! implementor rather than a degraded one — it has no method to stub, so it has no
//! `unimplemented!()` to panic from. Cypher lives in the separate [`Cypher`]
//! extension trait, advertised by [`GraphDb::cypher_support`]. A caller holding only
//! a `&dyn GraphDb` gets the honest typed [`GraphError::NoQueryLanguage`] from
//! [`no_cypher_here`].
//!
//! ## `open()` LISTS BEFORE IT PROBES, and that order is load bearing
//!
//! FalkorDB **creates** a graph key on the first `GRAPH.QUERY` against an unknown
//! name. Probing first therefore makes the probe itself the reason the answer is
//! "yes": the existence check could never fire, and a typo'd graph name would
//! silently leave a stray empty graph on the endpoint. Measured against a live
//! FalkorDB (korp lane `graph-trait`, 2026-08-31 — the probe-then-check order really
//! did add a `korp_absent_<pid>` key to `GRAPH.LIST`). Costs one extra round trip per
//! open; correctness wins. Guarded by `open_lists_before_it_probes`.
//!
//! ## Empty is not the same as unreachable
//!
//! Distinguished by TYPE, never by prose:
//!
//! * reachable, zero graphs ⇒ `Ok(vec![])`
//! * unreachable endpoint ⇒ [`GraphError::Connect`]
//! * missing graph ⇒ [`GraphError::NotFound`]
//! * past the time bound ⇒ [`GraphError::Timeout`]
//!
//! ## Result shape — [`SubGraph`]
//!
//! Every read returns a backend-neutral [`SubGraph`]: [`GraphNode`]s (a stable `id`,
//! a `label`, and the **raw** `natural_key` string a node carried, e.g. a Vessel's
//! `mmsi`) and id-referencing [`GraphEdge`]s. The consumer maps that onto its own
//! render types and resolves the natural key to an object identity — so no ontology
//! dependency leaks *into* this crate (which would form a cycle: this crate is a base
//! the consumer depends on).
//!
//! ## Colour is NOT here, deliberately
//!
//! A scene's colour legend stays with the renderer (korp's `graphstore::GraphScene`,
//! built from `korp::graphnav::label_rgb`). The hues are pinned to a consumer's own
//! **ontology** labels — `Person`, `Vessel`, `PoliceRecord` — and to the 2D/3D
//! navigators that paint them. Pulling the legend down here would drag that ontology
//! into a base crate and re-create exactly the cycle the `natural_key` design avoids,
//! and it would let a second consumer with different labels inherit the wrong palette.
//! What this crate owes the renderer is the labels; the renderer owns the colours.
//!
//! ## Dependency graph (no cycle)
//! ```text
//! korp ─▶ knut-graphdb ─▶ graphar-flight ─▶ redis (the ONE FalkorDB client)
//! └────▶ cozo (embedded Datalog)
//! ```
//! `knut-graphdb` lives in **knut** (alongside `graphar-flight`) precisely so it may
//! depend on `graphar-flight`; edda is a *base* others consume and must not.
//! **`knut → korp` is zero** and must stay zero.
use async_trait::async_trait;
mod types;
pub use types::{GraphEdge, GraphNode, GraphSchema, SubGraph, UpsertEdge, UpsertNode};
mod error;
pub use error::GraphError;
#[cfg(feature = "falkordb")]
pub mod falkor;
#[cfg(feature = "falkordb")]
pub use falkor::FalkorBackend;
#[cfg(feature = "cozo")]
pub mod cozo;
#[cfg(feature = "cozo")]
pub use cozo::CozoBackend;
/// The default row cap a `read`/`cypher` uses when a consumer passes `0`.
pub const DEFAULT_SCENE_LIMIT: usize = 1000;
/// `0` ⇒ [`DEFAULT_SCENE_LIMIT`]; anything else is the caller's own cap.
#[must_use]
pub fn cap(limit: usize) -> usize {
if limit == 0 { DEFAULT_SCENE_LIMIT } else { limit }
}
/// **What query language (if any) a backend offers.** Advertised by
/// [`GraphDb::cypher_support`] so a UI can hide a query bar a backend cannot serve,
/// instead of offering one that always errors.
#[derive(Clone, Copy, Debug, PartialEq, Eq, Default)]
pub enum CypherSupport {
/// No query language. [`GraphDb::graph_names`] / [`open`](GraphDb::open) /
/// [`read`](GraphDb::read) are the whole surface — and they are ENOUGH: this is a
/// complete, valid backend, not a degraded one.
#[default]
None,
/// openCypher, through the [`Cypher`] extension trait.
Cypher,
}
impl CypherSupport {
/// The stable machine key (state JSON / tests).
#[must_use]
pub fn key(self) -> &'static str {
match self {
CypherSupport::None => "none",
CypherSupport::Cypher => "cypher",
}
}
/// Does this backend answer [`Cypher::cypher`]?
#[must_use]
pub fn is_available(self) -> bool {
matches!(self, CypherSupport::Cypher)
}
}
/// A graph store a consumer can render, independent of the engine behind it.
///
/// Implementors provide **five** methods (`kind`, `endpoint`, `graph_names`, `open`,
/// `read`); `neighbours` / `shortest_path` / `upsert` / `cypher_support` are provided
/// with honest defaults, so the smallest complete backend is five functions long and
/// contains no stub that can panic.
///
/// Async because the FalkorDB backend is network-bound (the Cozo backend answers
/// immediately).
#[async_trait]
pub trait GraphDb: Send + Sync {
/// A short backend name for logs / introspection (`"falkordb"` / `"cozo"`).
fn kind(&self) -> &'static str;
/// The endpoint this store talks to (`redis://host:port`, a URL, a path —
/// whatever this backend's address is). Display + diagnostics only.
fn endpoint(&self) -> &str;
/// **Every graph on this endpoint**, sorted and de-duplicated.
///
/// `Ok(vec![])` means REACHABLE AND EMPTY. An unreachable endpoint MUST be
/// [`GraphError::Connect`] — never an empty list.
async fn graph_names(&self) -> Result<Vec<String>, GraphError>;
/// **Open a graph** — its labels, relationship types, property keys and
/// node/edge counts.
///
/// A graph that is not on this endpoint is [`GraphError::NotFound`], never an
/// empty [`GraphSchema`]. Implementors MUST establish existence from
/// [`graph_names`](GraphDb::graph_names) BEFORE probing the graph — see the crate
/// docs for why that order is load bearing.
async fn open(&self, graph: &str) -> Result<GraphSchema, GraphError>;
/// **Read a graph's nodes and edges**, capped at `limit` (0 ⇒
/// [`DEFAULT_SCENE_LIMIT`]).
///
/// NOTE the signature: a graph NAME and a cap. No query language. This is what
/// makes a Cypher-less backend a valid implementor rather than a special case.
async fn read(&self, graph: &str, limit: usize) -> Result<SubGraph, GraphError>;
/// **Neighbour expansion.** The 1-hop links of `node_id` (undirected), capped at
/// `limit` rows. The caller merges the result into the scene.
///
/// Provided default: [`GraphError::Unsupported`] — a backend that cannot expand
/// says so instead of returning a misleading empty sub-graph.
async fn neighbours(
&self,
graph: &str,
node_id: i64,
limit: usize,
) -> Result<SubGraph, GraphError> {
let _ = (graph, node_id, limit);
Err(GraphError::Unsupported {
backend: self.kind(),
op: "neighbours",
})
}
/// **Shortest path.** The bounded (≤ `max_hops`) shortest path between two nodes
/// as its own sub-graph. Provided default: [`GraphError::Unsupported`].
async fn shortest_path(
&self,
graph: &str,
from_id: i64,
to_id: i64,
max_hops: usize,
) -> Result<SubGraph, GraphError> {
let _ = (graph, from_id, to_id, max_hops);
Err(GraphError::Unsupported {
backend: self.kind(),
op: "shortest_path",
})
}
/// **Load / upsert.** Idempotently insert (or merge) `nodes` then `edges` into
/// `graph`. Returns the number of rows written. Provided default:
/// [`GraphError::Unsupported`] — a read-only backend is a valid backend.
async fn upsert(
&self,
graph: &str,
nodes: &[UpsertNode],
edges: &[UpsertEdge],
) -> Result<u64, GraphError> {
let _ = (graph, nodes, edges);
Err(GraphError::Unsupported {
backend: self.kind(),
op: "upsert",
})
}
/// What query language this backend offers. Default: [`CypherSupport::None`] — a
/// backend has to opt IN to advertising Cypher, so the honest answer is the one
/// you get by writing nothing.
fn cypher_support(&self) -> CypherSupport {
CypherSupport::None
}
}
/// **The OPTIONAL Cypher extension.** A backend that speaks openCypher implements
/// this IN ADDITION to [`GraphDb`] and overrides
/// [`cypher_support`](GraphDb::cypher_support) to say so.
///
/// It is a separate trait on purpose: a Cypher-less backend does not implement it at
/// all, so it has no stub to trip over and nothing to panic from.
#[async_trait]
pub trait Cypher: GraphDb {
/// Run one Cypher statement against `graph` and return the resulting sub-graph.
/// `limit` bounds the result (0 ⇒ [`DEFAULT_SCENE_LIMIT`]).
async fn cypher(
&self,
graph: &str,
query: &str,
limit: usize,
) -> Result<SubGraph, GraphError>;
}
/// **The honest refusal for a backend with no query language.** A UI that offers a
/// query bar over an arbitrary [`GraphDb`] calls this when the store does not
/// advertise Cypher: a typed [`GraphError::NoQueryLanguage`] naming the backend and
/// the endpoint — never a panic, never a silent empty result.
#[must_use]
pub fn no_cypher_here(store: &dyn GraphDb) -> GraphError {
GraphError::NoQueryLanguage {
backend: store.kind(),
endpoint: store.endpoint().to_string(),
}
}
/// The backend a consumer selected, parsed from the `KORP_GRAPH_BACKEND`
/// environment variable (`falkordb` — the default — or `cozo`).
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum BackendKind {
Falkordb,
Cozo,
}
impl BackendKind {
/// Read `KORP_GRAPH_BACKEND` (case-insensitive). Absent / empty / unknown ⇒
/// [`BackendKind::Falkordb`] (the default — nothing changes unless asked).
pub fn from_env() -> Self {
match std::env::var("KORP_GRAPH_BACKEND")
.ok()
.as_deref()
.map(str::trim)
.map(str::to_ascii_lowercase)
.as_deref()
{
Some("cozo") => BackendKind::Cozo,
_ => BackendKind::Falkordb,
}
}
}
#[cfg(test)]
mod tests {
use super::*;
/// **Gate 4 — Cypher is OPTIONAL, and the base trait proves it.**
///
/// `Pebble` implements `GraphDb` and NOTHING else: no `cypher`, no stub, no
/// `unimplemented!()`. It compiles, so a Cypher-less backend is a first-class
/// implementor.
///
/// RED when broken: move `cypher` onto `GraphDb` as a required method and this
/// impl stops compiling — the guard is the type checker, so it cannot be blind.
struct Pebble;
#[async_trait]
impl GraphDb for Pebble {
fn kind(&self) -> &'static str {
"pebble"
}
fn endpoint(&self) -> &str {
"mem://pebble"
}
async fn graph_names(&self) -> Result<Vec<String>, GraphError> {
Ok(vec!["only".to_string()])
}
async fn open(&self, graph: &str) -> Result<GraphSchema, GraphError> {
if graph != "only" {
return Err(GraphError::NotFound {
graph: graph.to_string(),
endpoint: self.endpoint().to_string(),
});
}
Ok(GraphSchema {
graph: graph.to_string(),
..Default::default()
})
}
async fn read(&self, _graph: &str, _limit: usize) -> Result<SubGraph, GraphError> {
Ok(SubGraph::default())
}
}
#[tokio::test]
async fn a_cypherless_backend_is_a_complete_implementor() {
let p = Pebble;
assert_eq!(p.cypher_support(), CypherSupport::None);
assert!(!p.cypher_support().is_available());
assert_eq!(p.cypher_support().key(), "none");
// The refusal is a VALUE, not a panic.
let refusal = no_cypher_here(&p);
assert!(matches!(
refusal,
GraphError::NoQueryLanguage {
backend: "pebble",
..
}
));
assert!(
refusal.to_string().contains("mem://pebble"),
"the refusal names the endpoint: {refusal}"
);
// And a missing graph is NotFound, not an empty schema.
let e = p.open("nope").await.unwrap_err();
assert!(matches!(e, GraphError::NotFound { .. }), "{e}");
// The provided defaults refuse honestly rather than returning empty scenes.
assert!(matches!(
p.neighbours("only", 1, 10).await.unwrap_err(),
GraphError::Unsupported {
backend: "pebble",
op: "neighbours"
}
));
assert!(matches!(
p.upsert("only", &[], &[]).await.unwrap_err(),
GraphError::Unsupported { .. }
));
}
/// `Pebble` must NOT satisfy `Cypher`. Proven by construction: a generic that
/// requires `Cypher` accepts a backend that implements it and — if the bound were
/// meaningless — would accept `Pebble` too.
#[test]
fn the_cypher_bound_really_bounds() {
fn needs_cypher<T: Cypher>(_: &T) {}
// Compile-time only; the negative half of this is the trybuild-free fact that
// `needs_cypher(&Pebble)` does not compile (asserted in the doc + by review).
let _ = needs_cypher::<DummyCypher>;
struct DummyCypher;
#[async_trait]
impl GraphDb for DummyCypher {
fn kind(&self) -> &'static str {
"dummy"
}
fn endpoint(&self) -> &str {
"mem://dummy"
}
async fn graph_names(&self) -> Result<Vec<String>, GraphError> {
Ok(Vec::new())
}
async fn open(&self, _g: &str) -> Result<GraphSchema, GraphError> {
Ok(GraphSchema::default())
}
async fn read(&self, _g: &str, _l: usize) -> Result<SubGraph, GraphError> {
Ok(SubGraph::default())
}
fn cypher_support(&self) -> CypherSupport {
CypherSupport::Cypher
}
}
#[async_trait]
impl Cypher for DummyCypher {
async fn cypher(
&self,
_g: &str,
_q: &str,
_l: usize,
) -> Result<SubGraph, GraphError> {
Ok(SubGraph::default())
}
}
}
#[test]
fn zero_limit_is_the_shared_default_cap() {
assert_eq!(cap(0), DEFAULT_SCENE_LIMIT);
assert_eq!(cap(0), 1000);
assert_eq!(cap(7), 7);
}
#[test]
fn a_schema_derived_from_a_scene_is_sorted_and_deduped() {
let sub = SubGraph {
nodes: vec![
GraphNode::new(1, "Vessel"),
GraphNode::new(2, "Person"),
GraphNode::new(3, "Person"),
GraphNode::new(4, ""),
],
edges: vec![
GraphEdge {
src_id: 1,
dst_id: 2,
rel: "OWNS".into(),
},
GraphEdge {
src_id: 2,
dst_id: 3,
rel: "OWNS".into(),
},
],
};
let s = GraphSchema::from_subgraph("g", &sub);
assert_eq!(s.labels, vec!["Person".to_string(), "Vessel".to_string()]);
assert_eq!(s.relationship_types, vec!["OWNS".to_string()]);
assert_eq!(s.node_count, 4);
assert_eq!(s.edge_count, 2);
}
}