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