khive_storage/graph.rs
1//! Graph storage capability — edge CRUD and traversal.
2
3use async_trait::async_trait;
4use khive_types::EdgeRelation;
5use uuid::Uuid;
6
7use crate::capability::StorageCapability;
8use crate::error::StorageError;
9use crate::types::{
10 BatchWriteSummary, DeleteMode, DirectedNeighborHit, Direction, Edge, EdgeFilter, EdgeSeekPage,
11 EdgeSortField, GraphPath, GuardedBatchOutcome, GuardedWriteOutcome, LinkId, NeighborHit,
12 NeighborQuery, Page, PageRequest, SortOrder, StorageResult, TraversalRequest,
13};
14
15/// Directed edge CRUD and graph traversal over the knowledge graph.
16#[async_trait]
17pub trait GraphStore: Send + Sync + 'static {
18 /// Insert or update a single edge.
19 async fn upsert_edge(&self, edge: Edge) -> StorageResult<()>;
20 /// Insert or update a batch of edges.
21 async fn upsert_edges(&self, edges: Vec<Edge>) -> StorageResult<BatchWriteSummary>;
22 /// Insert or update a single edge, re-checking that both endpoints still
23 /// exist (and are not soft-deleted) as part of the same write, not a
24 /// separate prior read. Closes the TOCTOU window between an async
25 /// prepare-time existence check and a later, unconditional write: a
26 /// concurrent hard-delete of an endpoint that lands between the two can
27 /// otherwise leave a durably dangling edge (#769).
28 ///
29 /// Returns [`GuardedWriteOutcome::Refused`] naming exactly which
30 /// endpoint(s) were missing, determined by the guard's own in-transaction
31 /// probe — never reconstructed by a caller re-reading the endpoints after
32 /// the write already failed, since a concurrent write landing between the
33 /// refusal and any such later read could misreport which endpoint was
34 /// actually missing at write time.
35 ///
36 /// Default returns `StorageError::Unsupported`: a backend that does not
37 /// override this method cannot honor the endpoint-existence guarantee,
38 /// and silently falling back to [`GraphStore::upsert_edge`] would
39 /// reintroduce the TOCTOU window this method exists to close.
40 async fn upsert_edge_guarded(&self, _edge: Edge) -> StorageResult<GuardedWriteOutcome> {
41 Err(StorageError::Unsupported {
42 capability: StorageCapability::Graph,
43 operation: "upsert_edge_guarded".into(),
44 message: "this backend does not implement guarded edge writes".into(),
45 })
46 }
47 /// Batch form of [`GraphStore::upsert_edge_guarded`]. All-or-nothing:
48 /// if any edge's endpoints are missing at write time, no edge from the
49 /// batch is persisted, `BatchWriteSummary::affected` is `0`, and
50 /// `GuardedBatchOutcome::refused` names the first failing batch entry and
51 /// its missing endpoint(s) — determined by the same in-transaction
52 /// pre-check that aborted the batch, not a post-hoc re-read.
53 ///
54 /// Default returns `StorageError::Unsupported`, for the same reason as
55 /// [`GraphStore::upsert_edge_guarded`]'s default.
56 async fn upsert_edges_guarded(&self, _edges: Vec<Edge>) -> StorageResult<GuardedBatchOutcome> {
57 Err(StorageError::Unsupported {
58 capability: StorageCapability::Graph,
59 operation: "upsert_edges_guarded".into(),
60 message: "this backend does not implement guarded edge writes".into(),
61 })
62 }
63 /// Fetch an edge by link ID, returning `None` if absent. Filters soft-deleted rows.
64 async fn get_edge(&self, id: LinkId) -> StorageResult<Option<Edge>>;
65 /// Fetch an edge by link ID including soft-deleted rows. Used by the runtime hard-delete path
66 /// to locate and namespace-check an already-soft-deleted edge before purging it.
67 async fn get_edge_including_deleted(&self, id: LinkId) -> StorageResult<Option<Edge>>;
68 /// Delete an edge by link ID using the specified delete mode.
69 async fn delete_edge(&self, id: LinkId, mode: DeleteMode) -> StorageResult<bool>;
70 /// Query edges with filter, sort, and pagination.
71 async fn query_edges(
72 &self,
73 filter: EdgeFilter,
74 sort: Vec<SortOrder<EdgeSortField>>,
75 page: PageRequest,
76 ) -> StorageResult<Page<Edge>>;
77 /// Count edges matching the given filter.
78 async fn count_edges(&self, filter: EdgeFilter) -> StorageResult<u64>;
79 /// Count edges grouped by relation, ignoring soft-deleted rows. Cheap
80 /// aggregate (`GROUP BY relation`) used to report the true per-relation
81 /// population for full-graph audits (#702.3).
82 async fn count_edges_by_relation(&self) -> StorageResult<Vec<(EdgeRelation, u64)>>;
83 /// Seek-pagination page of edges ordered by `id` ascending, using an
84 /// indexed range scan (`id > after`) against the `(namespace, id)`
85 /// primary key instead of `OFFSET`. `after` is exclusive; `None` starts
86 /// from the beginning of the set. Stable under concurrent writes and
87 /// O(log n + limit) at any depth, unlike offset paging (#702.2).
88 async fn query_edges_after(
89 &self,
90 filter: EdgeFilter,
91 after: Option<Uuid>,
92 limit: u32,
93 ) -> StorageResult<EdgeSeekPage>;
94 /// Return immediate neighbors of a graph node.
95 async fn neighbors(
96 &self,
97 node_id: Uuid,
98 query: NeighborQuery,
99 ) -> StorageResult<Vec<NeighborHit>>;
100 /// Return neighbors in BOTH directions in a single call, each tagged with
101 /// the direction (`Out`/`In`) it was found in. `query.direction` is
102 /// ignored — this always fetches both directions.
103 ///
104 /// Exists so a caller that needs both-direction neighbors labeled by
105 /// direction (e.g. the `context` verb) can do so with one storage query
106 /// instead of two separate direction-scoped `neighbors` calls. The
107 /// default implementation preserves the original two-call behavior for
108 /// backends that don't override it; `SqlGraphStore` overrides this with a
109 /// single `UNION ALL` query that projects a direction literal per arm.
110 async fn neighbors_both_directions(
111 &self,
112 node_id: Uuid,
113 query: NeighborQuery,
114 ) -> StorageResult<Vec<DirectedNeighborHit>> {
115 let mut out_query = query.clone();
116 out_query.direction = Direction::Out;
117 let mut in_query = query;
118 in_query.direction = Direction::In;
119 let mut result = Vec::new();
120 for hit in self.neighbors(node_id, out_query).await? {
121 result.push(DirectedNeighborHit {
122 hit,
123 direction: Direction::Out,
124 });
125 }
126 for hit in self.neighbors(node_id, in_query).await? {
127 result.push(DirectedNeighborHit {
128 hit,
129 direction: Direction::In,
130 });
131 }
132 Ok(result)
133 }
134 /// Fetch multiple edges by their link IDs in a single round-trip.
135 ///
136 /// IDs that are not found (absent or soft-deleted) are silently skipped;
137 /// the returned `Vec` may be shorter than `ids`. Backends that support
138 /// batched `IN (...)` queries should override this; the default loops
139 /// `get_edge` so non-SQLite backends keep compiling unchanged.
140 ///
141 /// Callers must chunk large ID lists before calling if they need a strict
142 /// size bound; this method does not enforce a maximum.
143 async fn get_edges(&self, ids: &[LinkId]) -> StorageResult<Vec<Edge>> {
144 let mut out = Vec::with_capacity(ids.len());
145 for &id in ids {
146 if let Some(edge) = self.get_edge(id).await? {
147 out.push(edge);
148 }
149 }
150 Ok(out)
151 }
152 /// Return neighbors for multiple source nodes in a single round-trip,
153 /// yielding `(source_id, hit)` pairs.
154 ///
155 /// The `query` parameters (direction, relations, min_weight) are applied
156 /// uniformly to every source node. `query.limit` is applied **per source**:
157 /// each source returns at most `limit` hits. Backends that support batched
158 /// `source_id IN (...)` queries should override this; the default loops
159 /// `neighbors` so non-SQLite backends keep compiling unchanged.
160 async fn batch_neighbors(
161 &self,
162 sources: &[Uuid],
163 query: NeighborQuery,
164 ) -> StorageResult<Vec<(Uuid, NeighborHit)>> {
165 let mut out = Vec::new();
166 for &src in sources {
167 let hits = self.neighbors(src, query.clone()).await?;
168 for hit in hits {
169 out.push((src, hit));
170 }
171 }
172 Ok(out)
173 }
174 /// Multi-hop BFS traversal from the given roots.
175 async fn traverse(&self, request: TraversalRequest) -> StorageResult<Vec<GraphPath>>;
176 /// Hard-delete every incident edge (source or target) for `node_id`, regardless of soft-delete
177 /// state. Used during endpoint hard-delete to prevent dangling `graph_edges` rows (ADR-002
178 /// no-dangling-references contract).
179 async fn purge_incident_edges(&self, node_id: Uuid) -> StorageResult<u64>;
180}