khive_storage/graph.rs
1//! Graph storage capability — edge CRUD and traversal.
2
3use async_trait::async_trait;
4use chrono::{DateTime, Utc};
5use khive_types::EdgeRelation;
6use uuid::Uuid;
7
8use crate::capability::StorageCapability;
9use crate::error::StorageError;
10use crate::types::{
11 BatchWriteSummary, DeleteMode, DirectedNeighborHit, Direction, Edge, EdgeEndpointBaseCounts,
12 EdgeFilter, EdgeSeekPage, EdgeSortField, EdgeUpsertRequest, EdgeUpsertResult, GraphPath,
13 GuardedBatchOutcome, GuardedEdgeBatchUpsertOutcome, GuardedEdgeUpsertOutcome,
14 GuardedWriteOutcome, LinkId, NeighborCursor, NeighborHit, NeighborQuery, Page, PageRequest,
15 SeekCursor, SeekPage, SortOrder, SqlStatement, StorageResult, TraversalRequest,
16};
17
18/// One caller-labelled predicate evaluated inside a graph writer transaction.
19///
20/// The statement must be one SELECT (including WITH followed by SELECT) returning
21/// exactly one row and one column. SQLite INTEGER 1 passes and INTEGER 0 refuses;
22/// other values and result shapes refuse with a typed reason. The caller owns
23/// the predicate and its tables. This is trusted in-process SQL, not a sandbox.
24#[derive(Clone, Debug)]
25pub struct GraphStatementGuard {
26 pub label: String,
27 pub statement: SqlStatement,
28}
29
30/// SQLite value class observed in a guard's non-boolean result.
31#[derive(Clone, Copy, Debug, PartialEq, Eq)]
32pub enum GraphGuardValueKind {
33 Null,
34 Integer,
35 Real,
36 Text,
37 Blob,
38}
39
40/// Predicate or result-shape refusal from one graph statement guard.
41///
42/// SQL preparation, binding and execution failures remain storage errors.
43#[derive(Clone, Debug, PartialEq, Eq)]
44pub enum GraphStatementGuardFailureReason {
45 PredicateFalse,
46 NoRows,
47 MultipleRows,
48 ColumnCount { actual: usize },
49 NonBoolean { kind: GraphGuardValueKind },
50}
51
52impl std::fmt::Display for GraphStatementGuardFailureReason {
53 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
54 match self {
55 Self::PredicateFalse => f.write_str("predicate_false"),
56 Self::NoRows => f.write_str("zero_rows"),
57 Self::MultipleRows => f.write_str("multiple_rows"),
58 Self::ColumnCount { actual } => write!(f, "column_count({actual})"),
59 Self::NonBoolean { kind } => write!(f, "non_boolean({kind:?})"),
60 }
61 }
62}
63
64/// The first refused guard, identified by its required caller label.
65///
66/// Display includes the label and reason, never the SQL text or bound values.
67#[derive(Clone, Debug, PartialEq, Eq)]
68pub struct GraphStatementGuardFailure {
69 pub label: String,
70 pub reason: GraphStatementGuardFailureReason,
71}
72
73impl std::fmt::Display for GraphStatementGuardFailure {
74 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
75 write!(
76 f,
77 "graph statement guard {:?} failed: {}",
78 self.label, self.reason
79 )
80 }
81}
82
83/// Transaction-observed result of a guarded, create-only graph insertion.
84///
85/// Existing rows are never replaced or resurrected. A natural-key winner takes
86/// precedence when the requested ID also collides; an unrelated ID collision
87/// without a natural-key winner remains a storage error. Only Created carries
88/// a new edge and permits its corresponding lifecycle event.
89#[derive(Clone, Debug)]
90pub enum EdgeInsertIfAbsentOutcome {
91 Created(Edge),
92 ExistingLive,
93 Tombstoned,
94 GuardFailed(GraphStatementGuardFailure),
95}
96
97/// The exact persisted cursor row observed by the reconciliation preview.
98#[derive(Clone, Debug)]
99pub struct CommitAnnotationCursorValue {
100 pub value: Vec<u8>,
101 pub updated_at: i64,
102}
103
104/// Preconditions carried from the selected repository and preview into the
105/// graph store's writer transaction.
106#[derive(Clone, Debug)]
107pub struct CommitAnnotationGuard {
108 pub expected_sha: String,
109 pub source_identity: String,
110 pub commits: CommitAnnotationCursorValue,
111 pub checkpoint: CommitAnnotationCursorValue,
112}
113
114/// Result of a create-only historical commit-to-project annotation attempt.
115/// The store decides this under its writer transaction; a caller-side preview
116/// is never authority to replace or resurrect an edge.
117#[derive(Clone, Debug)]
118pub enum CommitAnnotationInsertOutcome {
119 Created(Edge),
120 ExistingLive,
121 Tombstoned,
122 SourceChanged,
123 TargetChanged,
124 CursorChanged,
125}
126
127/// A validated symmetric-edge patch and the exact persisted snapshot it replaces.
128///
129/// The caller supplies canonical endpoints for a symmetric relation. `namespace`
130/// belongs to the stored record, independently of the store's routing namespace.
131/// Snapshot timestamps use raw microseconds so overflow is refused without a
132/// lossy timestamp conversion. Creation time, deletion state and target backend
133/// are not replaced by this request.
134#[derive(Clone, Debug)]
135pub struct SymmetricEdgeUpdateRequest {
136 pub namespace: String,
137 pub id: LinkId,
138 pub source_id: Uuid,
139 pub target_id: Uuid,
140 pub relation: EdgeRelation,
141 pub weight: f64,
142 pub metadata: Option<serde_json::Value>,
143 pub expected_updated_at_micros: i64,
144 pub expected_deleted_at_micros: Option<i64>,
145}
146
147/// Committed disposition of a guarded symmetric-edge update.
148#[derive(Debug)]
149pub enum SymmetricEdgeUpdateOutcome {
150 /// No competing natural key existed; the requested edge was updated.
151 Updated,
152 /// A live or tombstoned canonical row survived unchanged and the requested
153 /// duplicate was deleted. Its stored ID remains raw: parsing belongs after
154 /// commit, so malformed legacy data cannot change the transaction outcome.
155 Absorbed(String),
156 /// The requested row disappeared or no longer matched its revision or
157 /// deletion marker. Neither the update nor the conflict deletion occurred.
158 Stale,
159}
160
161/// Directed edge CRUD and graph traversal over the knowledge graph.
162#[async_trait]
163pub trait GraphStore: Send + Sync + 'static {
164 /// Return the newest live note of `kind` carrying the exact string `tag`
165 /// in its properties.tags array and connected to `node_id` by a live
166 /// incoming `annotates` edge in this store's namespace. Returns its UUID
167 /// and creation timestamp; equal timestamps choose the smallest UUID.
168 ///
169 /// Apply all predicates before limiting to one result. Note lookup follows
170 /// the by-ID contract; visibility is determined by the annotation edge's
171 /// namespace, not by introducing a second namespace filter on the note.
172 /// The note and edge must belong to this backend. Unsupported backends
173 /// fail explicitly rather than scanning an arbitrary annotation window.
174 async fn latest_annotating_note(
175 &self,
176 _node_id: Uuid,
177 _kind: &str,
178 _tag: &str,
179 ) -> StorageResult<Option<(Uuid, i64)>> {
180 Err(StorageError::Unsupported {
181 capability: StorageCapability::Graph,
182 operation: "latest_annotating_note".into(),
183 message: "this backend does not implement latest matching annotation lookup".into(),
184 })
185 }
186
187 /// Like `latest_annotating_note`, but also require one exact top-level
188 /// string property before selecting the newest candidate.
189 async fn latest_annotating_note_with_property(
190 &self,
191 _node_id: Uuid,
192 _kind: &str,
193 _tag: &str,
194 _property_key: &str,
195 _property_value: &str,
196 ) -> StorageResult<Option<(Uuid, i64)>> {
197 Err(StorageError::Unsupported {
198 capability: StorageCapability::Graph,
199 operation: "latest_annotating_note_with_property".into(),
200 message: "this backend does not implement latest matching annotation lookup".into(),
201 })
202 }
203
204 /// Insert or update a single edge.
205 async fn upsert_edge(&self, edge: Edge) -> StorageResult<()>;
206 /// Insert an edge only when neither its id nor natural key already
207 /// exists. Returns `true` when this call inserted the row and `false`
208 /// when an existing row won the race. The existing row is never updated.
209 ///
210 /// The default returns `Unsupported` rather than falling back to
211 /// [`GraphStore::upsert_edge`], because an upsert would overwrite the
212 /// winning row and violate this method's conditional-insert contract.
213 async fn insert_edge_if_absent(&self, _edge: Edge) -> StorageResult<bool> {
214 Err(StorageError::Unsupported {
215 capability: StorageCapability::Graph,
216 operation: "insert_edge_if_absent".into(),
217 message: "this backend does not implement conditional edge insert".into(),
218 })
219 }
220 /// Create an `annotates` edge only while its source is a live `commit`
221 /// note in `edge.namespace` with the exact SHA, its target is the live
222 /// project for the selected source, the paired commit cursor rows still
223 /// match the preview, and no row of the edge's natural key exists,
224 /// including a tombstone. All checks and the insert occur in one writer
225 /// transaction. Existing rows are immutable.
226 async fn insert_commit_annotation_if_absent(
227 &self,
228 _edge: Edge,
229 _guard: CommitAnnotationGuard,
230 ) -> StorageResult<CommitAnnotationInsertOutcome> {
231 Err(StorageError::Unsupported {
232 capability: StorageCapability::Graph,
233 operation: "insert_commit_annotation_if_absent".into(),
234 message: "this backend does not implement guarded commit annotation insert".into(),
235 })
236 }
237 /// Insert or update a batch of edges.
238 async fn upsert_edges(&self, edges: Vec<Edge>) -> StorageResult<BatchWriteSummary>;
239 /// Insert or replace one edge and return the transaction-observed
240 /// disposition plus preimage. Tombstone restoration is controlled by the
241 /// request rather than being an implicit side effect of every upsert.
242 async fn upsert_edge_observed(
243 &self,
244 _request: EdgeUpsertRequest,
245 ) -> StorageResult<EdgeUpsertResult> {
246 Err(StorageError::Unsupported {
247 capability: StorageCapability::Graph,
248 operation: "upsert_edge_observed".into(),
249 message: "this backend does not implement observed edge upserts".into(),
250 })
251 }
252 /// Replace an edge only when the persisted row still matches the
253 /// caller's read snapshot.
254 ///
255 /// `expected_updated_at` is the snapshot revision and
256 /// `expected_deleted_at` closes the soft-delete race. The replacement
257 /// edge's `updated_at` must be strictly greater than that persisted
258 /// revision. Returns `false` when the row disappeared, changed, or was
259 /// supplied a non-advancing replacement revision. This is the full-edge
260 /// compare-and-swap seam used when a caller derives coupled fields from
261 /// that snapshot before persistence — mirrors
262 /// [`crate::NoteStore::replace_note_if_unchanged`]. The default returns
263 /// `Unsupported` rather than falling back to an unguarded upsert and
264 /// reintroducing the stale-snapshot race.
265 async fn replace_edge_if_unchanged(
266 &self,
267 _edge: Edge,
268 _expected_updated_at: DateTime<Utc>,
269 _expected_deleted_at: Option<DateTime<Utc>>,
270 ) -> StorageResult<bool> {
271 Err(StorageError::Unsupported {
272 capability: StorageCapability::Graph,
273 operation: "replace_edge_if_unchanged".into(),
274 message: "this backend does not implement guarded edge replacement".into(),
275 })
276 }
277 /// Apply a validated symmetric-edge patch in one writer transaction.
278 ///
279 /// First require a representable revision strictly after the snapshot, even
280 /// if a competing natural key will absorb this edge. Then probe that key in
281 /// the request's record namespace, including tombstones. A competing row
282 /// survives unchanged; only the requested duplicate may be deleted. With no
283 /// competitor, replace the requested patch fields and advance its revision.
284 /// Both mutations compare the snapshot's revision and deletion marker.
285 ///
286 /// The default refuses rather than substituting an unguarded upsert or a
287 /// caller-side read/delete sequence. Errors can report unknown settlement;
288 /// callers must not infer rollback or retry an ambiguous result.
289 async fn update_symmetric_edge_if_unchanged(
290 &self,
291 _request: SymmetricEdgeUpdateRequest,
292 ) -> StorageResult<SymmetricEdgeUpdateOutcome> {
293 Err(StorageError::Unsupported {
294 capability: StorageCapability::Graph,
295 operation: "update_symmetric_edge_if_unchanged".into(),
296 message: "this backend does not implement guarded symmetric-edge updates".into(),
297 })
298 }
299 /// Insert or update a single edge, re-checking that both endpoints still
300 /// exist (and are not soft-deleted) as part of the same write, not a
301 /// separate prior read. Closes the TOCTOU window between an async
302 /// prepare-time existence check and a later, unconditional write: a
303 /// concurrent hard-delete of an endpoint that lands between the two can
304 /// otherwise leave a durably dangling edge (#769).
305 ///
306 /// Returns [`GuardedWriteOutcome::Refused`] naming exactly which
307 /// endpoint(s) were missing, determined by the guard's own in-transaction
308 /// probe — never reconstructed by a caller re-reading the endpoints after
309 /// the write already failed, since a concurrent write landing between the
310 /// refusal and any such later read could misreport which endpoint was
311 /// actually missing at write time.
312 ///
313 /// Default returns `StorageError::Unsupported`: a backend that does not
314 /// override this method cannot honor the endpoint-existence guarantee,
315 /// and silently falling back to [`GraphStore::upsert_edge`] would
316 /// reintroduce the TOCTOU window this method exists to close.
317 async fn upsert_edge_guarded(&self, _edge: Edge) -> StorageResult<GuardedWriteOutcome> {
318 Err(StorageError::Unsupported {
319 capability: StorageCapability::Graph,
320 operation: "upsert_edge_guarded".into(),
321 message: "this backend does not implement guarded edge writes".into(),
322 })
323 }
324 /// Observed form of [`GraphStore::upsert_edge_guarded`]. In addition to
325 /// the endpoint guard, it distinguishes create, live replacement, and
326 /// explicit resurrection without a caller-side read/write race.
327 async fn upsert_edge_guarded_observed(
328 &self,
329 _request: EdgeUpsertRequest,
330 ) -> StorageResult<GuardedEdgeUpsertOutcome> {
331 Err(StorageError::Unsupported {
332 capability: StorageCapability::Graph,
333 operation: "upsert_edge_guarded_observed".into(),
334 message: "this backend does not implement observed guarded edge upserts".into(),
335 })
336 }
337 /// Batch form of [`GraphStore::upsert_edge_guarded`]. All-or-nothing:
338 /// if any edge's endpoints are missing at write time, no edge from the
339 /// batch is persisted, `BatchWriteSummary::affected` is `0`, and
340 /// `GuardedBatchOutcome::refused` names the first failing batch entry and
341 /// its missing endpoint(s) — determined by the same in-transaction
342 /// pre-check that aborted the batch, not a post-hoc re-read.
343 /// Retain the original ordered input to enumerate every aborted write via
344 /// [`GuardedBatchOutcome::refusal_page`] beyond the default bounded sample.
345 ///
346 /// Default returns `StorageError::Unsupported`, for the same reason as
347 /// [`GraphStore::upsert_edge_guarded`]'s default.
348 async fn upsert_edges_guarded(&self, _edges: Vec<Edge>) -> StorageResult<GuardedBatchOutcome> {
349 Err(StorageError::Unsupported {
350 capability: StorageCapability::Graph,
351 operation: "upsert_edges_guarded".into(),
352 message: "this backend does not implement guarded edge writes".into(),
353 })
354 }
355 /// All-or-nothing observed batch form. Implementations must perform
356 /// endpoint and tombstone-policy preflight in the same write transaction
357 /// before applying any row.
358 async fn upsert_edges_guarded_observed(
359 &self,
360 _requests: Vec<EdgeUpsertRequest>,
361 ) -> StorageResult<GuardedEdgeBatchUpsertOutcome> {
362 Err(StorageError::Unsupported {
363 capability: StorageCapability::Graph,
364 operation: "upsert_edges_guarded_observed".into(),
365 message: "this backend does not implement observed guarded edge batches".into(),
366 })
367 }
368 /// Fetch an edge by link ID, returning `None` if absent. Filters soft-deleted rows.
369 async fn get_edge(&self, id: LinkId) -> StorageResult<Option<Edge>>;
370 /// Fetch an edge by link ID including soft-deleted rows. Used by the runtime hard-delete path
371 /// to locate and namespace-check an already-soft-deleted edge before purging it.
372 async fn get_edge_including_deleted(&self, id: LinkId) -> StorageResult<Option<Edge>>;
373 /// Fetch an edge by natural key (namespace, source, target, relation) including
374 /// soft-deleted rows. Used by the atomic-apply result renderer for a symmetric-relation
375 /// update whose surviving canonical row may be tombstoned (ADR-039 DO NOTHING) — the
376 /// normal `query_edges`/`list_edges` path filters `deleted_at IS NULL` and would report
377 /// "not found" for exactly that row.
378 ///
379 /// `namespace` is the natural key's own `namespace` column value (part of the
380 /// `UNIQUE(namespace, source_id, target_id, relation)` constraint this method queries by)
381 /// — it is passed explicitly rather than implied by whichever store instance `self` is,
382 /// so a caller who resolved the record's namespace independently of its own ambient token
383 /// (the atomic-apply renderer, which knows the committed edge's namespace from its prepare-
384 /// time `EdgeNaturalKey`, not from the caller's token) cannot accidentally query the wrong
385 /// namespace by relying on implicit store scoping.
386 async fn get_edge_by_natural_key_including_deleted(
387 &self,
388 namespace: &str,
389 source_id: Uuid,
390 target_id: Uuid,
391 relation: EdgeRelation,
392 ) -> StorageResult<Option<Edge>>;
393 /// Delete an edge by link ID using the specified delete mode.
394 async fn delete_edge(&self, id: LinkId, mode: DeleteMode) -> StorageResult<bool>;
395 /// Query edges with filter, sort, and pagination without an implicit
396 /// exact count. Implementations should return `total: None`; callers that
397 /// need a count use [`Self::count_edges`] explicitly.
398 async fn query_edges(
399 &self,
400 filter: EdgeFilter,
401 sort: Vec<SortOrder<EdgeSortField>>,
402 page: PageRequest,
403 ) -> StorageResult<Page<Edge>>;
404 /// Query edges across the given namespaces in one deterministic query
405 /// with real SQL paging. The multi-namespace analogue of
406 /// [`Self::query_edges`]: a single statement with `namespace IN (...)`
407 /// keeps `offset` continuation coherent, where fetching per-namespace
408 /// prefixes and slicing a client-side merge floats the window between
409 /// calls (silent duplicate/skip enumeration). Backends without batched
410 /// namespace support retain the single-namespace path and reject
411 /// multi-namespace requests explicitly. Implementations should return
412 /// `total: None`; callers that need a count use
413 /// [`Self::count_edges_in_namespaces`] explicitly.
414 async fn query_edges_in_namespaces(
415 &self,
416 namespaces: &[String],
417 filter: EdgeFilter,
418 sort: Vec<SortOrder<EdgeSortField>>,
419 page: PageRequest,
420 ) -> StorageResult<Page<Edge>> {
421 match namespaces.len() {
422 0 => Ok(Page {
423 items: Vec::new(),
424 total: None,
425 }),
426 1 => self.query_edges(filter, sort, page).await,
427 _ => Err(StorageError::Unsupported {
428 capability: StorageCapability::Graph,
429 operation: "query_edges_in_namespaces".into(),
430 message: "this backend does not implement batched namespace edge queries".into(),
431 }),
432 }
433 }
434 /// Count edges matching the given filter.
435 async fn count_edges(&self, filter: EdgeFilter) -> StorageResult<u64>;
436 /// Count edges across the given namespaces in one aggregate query.
437 /// Backends without batched namespace support retain the single-namespace
438 /// path and reject multi-namespace requests explicitly.
439 async fn count_edges_in_namespaces(
440 &self,
441 namespaces: &[String],
442 filter: EdgeFilter,
443 ) -> StorageResult<u64> {
444 match namespaces.len() {
445 0 => Ok(0),
446 1 => self.count_edges(filter).await,
447 _ => Err(StorageError::Unsupported {
448 capability: StorageCapability::Graph,
449 operation: "count_edges_in_namespaces".into(),
450 message: "this backend does not implement batched namespace edge counts".into(),
451 }),
452 }
453 }
454 /// Count edges grouped by relation, ignoring soft-deleted rows. Cheap
455 /// aggregate (`GROUP BY relation`) used to report the true per-relation
456 /// population for full-graph audits (#702.3).
457 async fn count_edges_by_relation(&self) -> StorageResult<Vec<(EdgeRelation, u64)>>;
458 /// Count edges grouped by relation across the given namespaces in one
459 /// aggregate query.
460 async fn count_edges_by_relation_in_namespaces(
461 &self,
462 namespaces: &[String],
463 ) -> StorageResult<Vec<(EdgeRelation, u64)>> {
464 match namespaces.len() {
465 0 => Ok(Vec::new()),
466 1 => self.count_edges_by_relation().await,
467 _ => Err(StorageError::Unsupported {
468 capability: StorageCapability::Graph,
469 operation: "count_edges_by_relation_in_namespaces".into(),
470 message: "this backend does not implement batched namespace relation counts".into(),
471 }),
472 }
473 }
474 /// Count live edges grouped by the base each endpoint resolves against.
475 ///
476 /// A relation breakdown cannot answer this: relations do not determine
477 /// endpoint bases, and the same relation appears on both sides of the
478 /// structure/provenance line. One aggregate query, same shape and cost as
479 /// the relation counts.
480 async fn count_edges_by_endpoint_base(&self) -> StorageResult<EdgeEndpointBaseCounts> {
481 Err(StorageError::Unsupported {
482 capability: StorageCapability::Graph,
483 operation: "count_edges_by_endpoint_base".into(),
484 message: "this backend does not implement endpoint-base edge counts".into(),
485 })
486 }
487 /// Count live edges by endpoint base across the given namespaces in one
488 /// aggregate query.
489 async fn count_edges_by_endpoint_base_in_namespaces(
490 &self,
491 namespaces: &[String],
492 ) -> StorageResult<EdgeEndpointBaseCounts> {
493 match namespaces.len() {
494 0 => Ok(EdgeEndpointBaseCounts::default()),
495 1 => self.count_edges_by_endpoint_base().await,
496 _ => Err(StorageError::Unsupported {
497 capability: StorageCapability::Graph,
498 operation: "count_edges_by_endpoint_base_in_namespaces".into(),
499 message: "this backend does not implement batched namespace endpoint-base counts"
500 .into(),
501 }),
502 }
503 }
504 /// Count live edges across the given namespaces grouped by relation and,
505 /// within each relation, by the base each endpoint resolves against.
506 ///
507 /// The live edge total, the relation counts and the endpoint-base counts
508 /// describe one population, so one grouped scan answers all three:
509 /// summing each relation's buckets gives its relation count, and summing
510 /// the buckets across relations gives the endpoint-base counts. Asking for
511 /// them as three aggregates walks the edge table and its endpoint probes
512 /// three times. Backends without this aggregate report `Unsupported`, and
513 /// callers fall back to the three single-purpose counts.
514 async fn count_edges_by_relation_and_endpoint_base_in_namespaces(
515 &self,
516 namespaces: &[String],
517 ) -> StorageResult<Vec<(EdgeRelation, EdgeEndpointBaseCounts)>> {
518 let _ = namespaces;
519 Err(StorageError::Unsupported {
520 capability: StorageCapability::Graph,
521 operation: "count_edges_by_relation_and_endpoint_base_in_namespaces".into(),
522 message: "this backend does not implement the combined relation and endpoint-base \
523 edge count"
524 .into(),
525 })
526 }
527 /// Seek-pagination page of edges ordered by `id` ascending, using an
528 /// indexed range scan (`id > after`) against the `(namespace, id)`
529 /// primary key instead of `OFFSET`. `after` is exclusive; `None` starts
530 /// from the beginning of the set. This remains an efficient compatibility
531 /// path for a fixed edge set, but random UUIDs inserted concurrently may
532 /// sort behind an issued boundary. Public concurrent walks use
533 /// [`Self::query_edges_sequence_after`] instead (#1424).
534 async fn query_edges_after(
535 &self,
536 filter: EdgeFilter,
537 after: Option<Uuid>,
538 limit: u32,
539 ) -> StorageResult<EdgeSeekPage>;
540 /// Resolve an edge id to its immutable insertion sequence.
541 async fn edge_sequence(&self, _id: Uuid) -> StorageResult<Option<i64>> {
542 Err(StorageError::Unsupported {
543 capability: StorageCapability::Graph,
544 operation: "edge_sequence".into(),
545 message: "this backend does not implement edge insertion sequences".into(),
546 })
547 }
548 /// Resolve edge ids to immutable insertion sequences. Implementations may
549 /// override this to batch the lookup; the default preserves correctness.
550 async fn edge_sequences(&self, ids: &[Uuid]) -> StorageResult<Vec<(Uuid, i64)>> {
551 let mut resolved = Vec::with_capacity(ids.len());
552 for id in ids {
553 if let Some(sequence) = self.edge_sequence(*id).await? {
554 resolved.push((*id, sequence));
555 }
556 }
557 Ok(resolved)
558 }
559 /// Seek-pagination page ordered by immutable insertion sequence. This is
560 /// the stable public-list contract for walks overlapping inserts (#1424).
561 async fn query_edges_sequence_after(
562 &self,
563 _filter: EdgeFilter,
564 _after: Option<SeekCursor>,
565 _limit: u32,
566 ) -> StorageResult<SeekPage<Edge>> {
567 Err(StorageError::Unsupported {
568 capability: StorageCapability::Graph,
569 operation: "query_edges_sequence_after".into(),
570 message: "this backend does not implement insertion-sequence edge pagination".into(),
571 })
572 }
573 /// Return immediate neighbors of a graph node.
574 async fn neighbors(
575 &self,
576 node_id: Uuid,
577 query: NeighborQuery,
578 ) -> StorageResult<Vec<NeighborHit>>;
579 /// Return one deterministic neighbor page. `after` is exclusive and
580 /// `neighbor_kinds`, when present, filters entity and note kinds before
581 /// the limit is applied. Backends that do not implement kind-aware paging
582 /// retain an explicit unsupported result rather than silently returning a
583 /// misleading page.
584 async fn neighbors_page(
585 &self,
586 node_id: Uuid,
587 mut query: NeighborQuery,
588 after: Option<NeighborCursor>,
589 neighbor_kinds: Option<Vec<String>>,
590 ) -> StorageResult<Vec<NeighborHit>> {
591 if neighbor_kinds
592 .as_ref()
593 .is_some_and(|kinds| !kinds.is_empty())
594 {
595 return Err(StorageError::Unsupported {
596 capability: StorageCapability::Graph,
597 operation: "neighbors_page".into(),
598 message: "this backend does not implement neighbor kind filtering".into(),
599 });
600 }
601 let limit = query.limit;
602 query.limit = None;
603 let mut hits = self.neighbors(node_id, query).await?;
604 if let Some(cursor) = after {
605 hits.retain(|hit| cursor.is_after(hit));
606 }
607 if let Some(limit) = limit {
608 hits.truncate(limit as usize);
609 }
610 Ok(hits)
611 }
612 /// Return neighbors in BOTH directions in a single call, each tagged with
613 /// the direction (`Out`/`In`) it was found in. `query.direction` is
614 /// ignored — this always fetches both directions.
615 ///
616 /// Exists so a caller that needs both-direction neighbors labeled by
617 /// direction (e.g. the `context` verb) can do so with one storage query
618 /// instead of two separate direction-scoped `neighbors` calls. The
619 /// default implementation preserves the original two-call behavior for
620 /// backends that don't override it; `SqlGraphStore` overrides this with a
621 /// single `UNION ALL` query that projects a direction literal per arm.
622 async fn neighbors_both_directions(
623 &self,
624 node_id: Uuid,
625 query: NeighborQuery,
626 ) -> StorageResult<Vec<DirectedNeighborHit>> {
627 let mut out_query = query.clone();
628 out_query.direction = Direction::Out;
629 let mut in_query = query;
630 in_query.direction = Direction::In;
631 let mut result = Vec::new();
632 for hit in self.neighbors(node_id, out_query).await? {
633 result.push(DirectedNeighborHit {
634 hit,
635 direction: Direction::Out,
636 });
637 }
638 for hit in self.neighbors(node_id, in_query).await? {
639 result.push(DirectedNeighborHit {
640 hit,
641 direction: Direction::In,
642 });
643 }
644 Ok(result)
645 }
646 /// Fetch multiple edges by their link IDs in a single round-trip.
647 ///
648 /// IDs that are not found (absent or soft-deleted) are silently skipped;
649 /// the returned `Vec` may be shorter than `ids`. Backends that support
650 /// batched `IN (...)` queries should override this; the default loops
651 /// `get_edge` so non-SQLite backends keep compiling unchanged.
652 ///
653 /// Callers must chunk large ID lists before calling if they need a strict
654 /// size bound; this method does not enforce a maximum.
655 async fn get_edges(&self, ids: &[LinkId]) -> StorageResult<Vec<Edge>> {
656 let mut out = Vec::with_capacity(ids.len());
657 for &id in ids {
658 if let Some(edge) = self.get_edge(id).await? {
659 out.push(edge);
660 }
661 }
662 Ok(out)
663 }
664 /// Read live edges in input order, retaining each row's decode outcome.
665 ///
666 /// The vector has exactly one entry per input, including duplicates and
667 /// missing/soft-deleted IDs (`Ok(None)`). An inner error belongs to that
668 /// input; an outer error is a statement/admission failure for the batch.
669 /// Callers must bound batches and select errors in their original order.
670 /// The default preserves backend support with point reads; SQLite batches.
671 async fn get_edge_read_outcomes(
672 &self,
673 ids: &[LinkId],
674 ) -> StorageResult<Vec<StorageResult<Option<Edge>>>> {
675 let mut outcomes = Vec::with_capacity(ids.len());
676 for &id in ids {
677 outcomes.push(self.get_edge(id).await);
678 }
679 Ok(outcomes)
680 }
681
682 /// Return neighbors for multiple source nodes in a single round-trip,
683 /// yielding `(source_id, hit)` pairs.
684 ///
685 /// The `query` parameters (direction, relations, min_weight) are applied
686 /// uniformly to every source node. `query.limit` is applied **per source**:
687 /// each source returns at most `limit` hits. Backends that support batched
688 /// `source_id IN (...)` queries should override this; the default loops
689 /// `neighbors` so non-SQLite backends keep compiling unchanged.
690 async fn batch_neighbors(
691 &self,
692 sources: &[Uuid],
693 query: NeighborQuery,
694 ) -> StorageResult<Vec<(Uuid, NeighborHit)>> {
695 let mut out = Vec::new();
696 for &src in sources {
697 let hits = self.neighbors(src, query.clone()).await?;
698 for hit in hits {
699 out.push((src, hit));
700 }
701 }
702 Ok(out)
703 }
704 /// Bounded multi-hop BFS traversal from the given roots.
705 ///
706 /// Implementations must validate [`TraversalRequest::validate`], count
707 /// adjacency rows before first-visit de-duplication against the request's
708 /// shared execution budget, stop a root as soon as its effective result
709 /// limit is filled, and return an error rather than partial paths when the
710 /// work or time budget expires. Minimum-depth BFS selection is normative;
711 /// same-depth tie ordering is not.
712 async fn traverse(&self, request: TraversalRequest) -> StorageResult<Vec<GraphPath>>;
713 /// Hard-delete every incident edge (source or target) for `node_id`, regardless of soft-delete
714 /// state. Used during endpoint hard-delete to prevent dangling `graph_edges` rows (ADR-002
715 /// no-dangling-references contract).
716 async fn purge_incident_edges(&self, node_id: Uuid) -> StorageResult<u64>;
717}