khive_runtime/operations/
record_ops.rs

1use khive_storage::note::Note;
2use khive_storage::types::{DeleteMode, Page, PageRequest, SeekCursor};
3use khive_storage::{Entity, EntityFilter, Event, EventFilter};
4use khive_types::KhiveError;
5use uuid::Uuid;
6
7#[cfg(doc)]
8use super::EntityCreateSpec;
9#[cfg(any(test, feature = "fault-injection"))]
10use super::{consume_fault, ENTITY_COMPENSATION_FAIL_NS};
11use super::{entity_stats_counts, EntityStatsCounts};
12use crate::atomic_runner::AtomicOpPlan;
13use crate::error::{RuntimeError, RuntimeResult};
14use crate::runtime::{KhiveRuntime, NamespaceToken};
15
16/// One note item for [`KhiveRuntime::prepare_bulk_note_plan`], the note
17/// analogue of [`EntityCreateSpec`]. The caller has already run the kind's
18/// own preparation hook, so owner-kind policy such as the memory pack's
19/// creation refusal happens before this point.
20#[derive(Clone, Debug)]
21pub struct NoteCreateSpec {
22    pub kind: String,
23    pub name: Option<String>,
24    pub content: String,
25    pub salience: Option<f64>,
26    pub properties: Option<serde_json::Value>,
27}
28
29impl KhiveRuntime {
30    // ---- Entity operations ----
31
32    pub(super) async fn compensate_entity_create(
33        &self,
34        token: &NamespaceToken,
35        entity_id: Uuid,
36        namespace: &str,
37        vector_models: &[String],
38    ) -> Vec<String> {
39        let mut cleanup_errors = Vec::new();
40
41        #[cfg(any(test, feature = "fault-injection"))]
42        let entity_delete_injected = consume_fault(&ENTITY_COMPENSATION_FAIL_NS, namespace);
43        #[cfg(not(any(test, feature = "fault-injection")))]
44        let entity_delete_injected = false;
45
46        if entity_delete_injected {
47            cleanup_errors.push("entity row delete: injected compensation failure".to_string());
48        } else {
49            match self.entities(token) {
50                Ok(store) => {
51                    if let Err(error) = store.delete_entity(entity_id, DeleteMode::Hard).await {
52                        cleanup_errors.push(format!("entity row delete: {error}"));
53                    }
54                }
55                Err(error) => cleanup_errors.push(format!("entity store access: {error}")),
56            }
57        }
58
59        match self.text(token) {
60            Ok(fts) => {
61                if let Err(error) = fts.delete_document(namespace, entity_id).await {
62                    cleanup_errors.push(format!("FTS document delete: {error}"));
63                }
64            }
65            Err(error) => cleanup_errors.push(format!("FTS store access: {error}")),
66        }
67
68        for model_name in vector_models {
69            match self.vectors_for_model(token, model_name) {
70                Ok(vectors) => {
71                    if let Err(error) = vectors.delete(entity_id).await {
72                        cleanup_errors
73                            .push(format!("vector delete for model {model_name}: {error}"));
74                    }
75                }
76                Err(error) => cleanup_errors.push(format!(
77                    "vector store access for model {model_name}: {error}"
78                )),
79            }
80        }
81
82        cleanup_errors
83    }
84
85    pub(super) fn entity_create_failure(
86        entity_id: Uuid,
87        primary: RuntimeError,
88        cleanup_errors: Vec<String>,
89    ) -> RuntimeError {
90        if cleanup_errors.is_empty() {
91            primary
92        } else {
93            RuntimeError::Khive(KhiveError::internal(format!(
94                "create_entity indexing failed for record {entity_id}; primary failure: \
95                 {primary}; compensation failure(s): {}; partial persistence is possible; \
96                 inspect and reconcile this record before retrying",
97                cleanup_errors.join("; ")
98            )))
99        }
100    }
101
102    /// Retrieve an entity by ID.
103    ///
104    /// UUID v4 is globally unique: no namespace filter on by-ID ops.
105    ///
106    /// Interim identifier-continuity disclosure (precedes the full transitive
107    /// redirect chase): a miss is probed once against the tombstone row. If
108    /// the id was consumed by `merge(into_id, from_id)` — `merged_into` set —
109    /// the `NotFound` message names the kept id so the caller can requery it
110    /// directly. Single-level only: it does not chase a chain of merges and
111    /// does not return the kept entity in place of the miss. The probe only
112    /// runs after the live-row lookup misses, so the happy path pays no
113    /// extra query.
114    pub async fn get_entity(&self, token: &NamespaceToken, id: Uuid) -> RuntimeResult<Entity> {
115        let store = self.entities(token)?;
116        if let Some(entity) = store.get_entity(id).await? {
117            return Ok(entity);
118        }
119        if let Some(tombstone) = store.get_entity_including_deleted(id).await? {
120            if let Some(kept_id) = tombstone.merged_into {
121                return Err(RuntimeError::NotFound(format!(
122                    "{id} was merged into {kept_id}; query the kept id"
123                )));
124            }
125        }
126        Err(RuntimeError::NotFound(format!("entity {id}")))
127    }
128
129    /// Retrieve an entity by ID including soft-deleted rows.
130    ///
131    /// UUID v4 is globally unique: no namespace filter on by-ID ops.
132    pub async fn get_entity_including_deleted(
133        &self,
134        token: &NamespaceToken,
135        id: Uuid,
136    ) -> RuntimeResult<Option<Entity>> {
137        self.entities(token)?
138            .get_entity_including_deleted(id)
139            .await
140            .map_err(Into::into)
141    }
142
143    /// Retrieve a note by ID including soft-deleted rows.
144    ///
145    /// UUID v4 is globally unique: no namespace filter on by-ID ops.
146    pub async fn get_note_including_deleted(
147        &self,
148        token: &NamespaceToken,
149        id: Uuid,
150    ) -> RuntimeResult<Option<khive_storage::note::Note>> {
151        self.notes(token)?
152            .get_note_including_deleted(id)
153            .await
154            .map_err(Into::into)
155    }
156
157    /// Fetch multiple entities by ID, returning only those that exist in the
158    /// caller's namespace.  Missing or namespace-mismatched IDs are silently
159    /// omitted so that batch lookups don't abort on a single stale reference.
160    pub async fn get_entities_by_ids(
161        &self,
162        token: &NamespaceToken,
163        ids: &[Uuid],
164    ) -> RuntimeResult<Vec<Entity>> {
165        if ids.is_empty() {
166            return Ok(vec![]);
167        }
168        let filter = EntityFilter {
169            ids: ids.to_vec(),
170            ..Default::default()
171        };
172        let page = self
173            .entities(token)?
174            .query_entities(
175                token.namespace().as_str(),
176                filter,
177                PageRequest {
178                    offset: 0,
179                    limit: ids.len() as u32,
180                },
181            )
182            .await?;
183        Ok(page.items)
184    }
185
186    /// Like `get_entities_by_ids` but scoped to the token's full visible-namespace
187    /// set (`primary ∪ extra_visible`) instead of primary only.
188    ///
189    /// Graph expansion (`neighbors`, `traverse`) iterates over all visible
190    /// namespaces, so enrichment must use the same scope — otherwise neighbors
191    /// or path nodes whose entities live in an extra-visible namespace are left
192    /// with `name = None`, `kind = None`.  Missing or out-of-scope IDs are
193    /// silently omitted (best-effort, same as `get_entities_by_ids`).
194    pub(super) async fn get_entities_by_ids_visible(
195        &self,
196        token: &NamespaceToken,
197        ids: &[Uuid],
198    ) -> RuntimeResult<Vec<Entity>> {
199        if ids.is_empty() {
200            return Ok(vec![]);
201        }
202        let namespaces: Vec<String> = token
203            .visible_namespaces()
204            .iter()
205            .map(|ns| ns.as_str().to_owned())
206            .collect();
207        let filter = EntityFilter {
208            ids: ids.to_vec(),
209            namespaces,
210            ..Default::default()
211        };
212        let page = self
213            .entities(token)?
214            .query_entities(
215                token.namespace().as_str(),
216                filter,
217                PageRequest {
218                    offset: 0,
219                    limit: ids.len() as u32,
220                },
221            )
222            .await?;
223        Ok(page.items)
224    }
225
226    /// Enforce that `record_ns` is within the caller's visible namespace set.
227    ///
228    /// Returns `Err(NotFound)` when the record namespace is not in the visible
229    /// set — wrong-namespace and absent UUIDs must be indistinguishable
230    /// externally (no existence oracle).
231    ///
232    /// When the visible set is a single entry equal to `caller_primary_ns`, this
233    /// is identical to the former strict-equality check (backward-compatible).
234    pub(crate) fn ensure_namespace(record_ns: &str, caller_primary_ns: &str) -> RuntimeResult<()> {
235        if record_ns == caller_primary_ns {
236            return Ok(());
237        }
238        Err(RuntimeError::NotFound("not found in this namespace".into()))
239    }
240
241    /// Enforce that `record_ns` is a member of the token's visible namespace set.
242    ///
243    /// This is the multi-namespace-aware variant used when the token carries an
244    /// extended visibility set. For single-namespace tokens (visible == [primary])
245    /// this degenerates to the same strict-equality check as `ensure_namespace`.
246    pub(crate) fn ensure_namespace_visible(
247        record_ns: &str,
248        token: &NamespaceToken,
249    ) -> RuntimeResult<()> {
250        for ns in token.visible_namespaces() {
251            if record_ns == ns.as_str() {
252                return Ok(());
253            }
254        }
255        Err(RuntimeError::NotFound("not found in this namespace".into()))
256    }
257
258    /// List entities visible to the token, optionally filtered by kind and entity_type.
259    /// A null entity_type falls back to a string properties.type for filtering only.
260    ///
261    /// When the token carries a multi-namespace visible set, entities from all
262    /// visible namespaces are returned. When the visible set is `[primary]`
263    /// (the default) this behaves identically to the pre-visibility behaviour.
264    pub async fn list_entities(
265        &self,
266        token: &NamespaceToken,
267        kind: Option<&str>,
268        entity_type: Option<&str>,
269        limit: u32,
270        offset: u32,
271    ) -> RuntimeResult<Vec<Entity>> {
272        let filter = EntityFilter {
273            kinds: kind
274                .map(|value| vec![value.to_string()])
275                .unwrap_or_default(),
276            entity_types: entity_type
277                .map(|value| vec![value.to_string()])
278                .unwrap_or_default(),
279            legacy_entity_type_fallback: true,
280            ..Default::default()
281        };
282        self.list_entities_filtered(token, filter, limit, offset)
283            .await
284    }
285
286    /// Apply a composed entity predicate before offset pagination. Namespace
287    /// visibility is supplied by the token, just as for the scalar list API.
288    pub async fn list_entities_filtered(
289        &self,
290        token: &NamespaceToken,
291        mut filter: EntityFilter,
292        limit: u32,
293        offset: u32,
294    ) -> RuntimeResult<Vec<Entity>> {
295        filter.namespaces = token
296            .visible_namespaces()
297            .iter()
298            .map(|namespace| namespace.as_str().to_owned())
299            .collect();
300        let page = self
301            .entities(token)?
302            .query_entities_count_free(
303                token.namespace().as_str(),
304                filter,
305                PageRequest {
306                    offset: offset.into(),
307                    limit,
308                },
309            )
310            .await?;
311        Ok(page.items)
312    }
313
314    /// List an immutable insertion-sequence page of visible entities.
315    ///
316    /// The public cursor remains the UUID of the last returned entity. We
317    /// resolve its immutable database-assigned sequence before querying so callers do
318    /// not need to serialize storage details. A missing or out-of-scope cursor
319    /// fails explicitly instead of silently resuming from the wrong boundary.
320    pub async fn list_entities_after(
321        &self,
322        token: &NamespaceToken,
323        kind: Option<&str>,
324        entity_type: Option<&str>,
325        tags_any: &[String],
326        after: Option<Uuid>,
327        limit: u32,
328    ) -> RuntimeResult<(Vec<Entity>, Option<Uuid>)> {
329        let filter = EntityFilter {
330            kinds: kind
331                .map(|value| vec![value.to_string()])
332                .unwrap_or_default(),
333            entity_types: entity_type
334                .map(|value| vec![value.to_string()])
335                .unwrap_or_default(),
336            legacy_entity_type_fallback: true,
337            tags_any: tags_any.to_vec(),
338            ..Default::default()
339        };
340        self.list_entities_after_filtered(token, filter, after, limit)
341            .await
342    }
343
344    /// Apply a composed entity predicate before insertion-sequence pagination,
345    /// preserving the scalar API's cursor validation and token visibility.
346    pub async fn list_entities_after_filtered(
347        &self,
348        token: &NamespaceToken,
349        mut filter: EntityFilter,
350        after: Option<Uuid>,
351        limit: u32,
352    ) -> RuntimeResult<(Vec<Entity>, Option<Uuid>)> {
353        let store = self.entities(token)?;
354        let after = match after {
355            Some(id) => {
356                let entity = self
357                    .get_entity_including_deleted(token, id)
358                    .await?
359                    .ok_or_else(|| RuntimeError::NotFound(format!("entity cursor {id}")))?;
360                Self::ensure_namespace_visible(&entity.namespace, token)?;
361                let sequence = store.entity_sequence(id).await?.ok_or_else(|| {
362                    RuntimeError::Internal(format!(
363                        "entity cursor {id} has no insertion-sequence ledger row"
364                    ))
365                })?;
366                Some(SeekCursor { sequence, id })
367            }
368            None => None,
369        };
370        filter.namespaces = token
371            .visible_namespaces()
372            .iter()
373            .map(|namespace| namespace.as_str().to_owned())
374            .collect();
375        let page = store
376            .query_entities_after(token.namespace().as_str(), filter, after, limit)
377            .await?;
378        Ok((page.items, page.next_after.map(|cursor| cursor.id)))
379    }
380
381    /// List entities filtered by kind, optional domain tag, limit, and offset.
382    ///
383    /// When `domain_tag` is Some, the query is restricted at the storage layer via
384    /// `EntityFilter::tags_any` so the page result already reflects the domain
385    /// constraint.  This avoids the silent truncation that occurs when filtering
386    /// post-page (K-3). Multi-namespace visibility from the token is applied.
387    pub async fn list_entities_tagged(
388        &self,
389        token: &NamespaceToken,
390        kind: Option<&str>,
391        domain_tag: Option<&str>,
392        limit: u32,
393        offset: u32,
394    ) -> RuntimeResult<Vec<Entity>> {
395        let ns_strs: Vec<String> = token
396            .visible_namespaces()
397            .iter()
398            .map(|ns| ns.as_str().to_owned())
399            .collect();
400        let filter = EntityFilter {
401            kinds: match kind {
402                Some(k) => vec![k.to_string()],
403                None => vec![],
404            },
405            tags_any: match domain_tag {
406                Some(t) if !t.is_empty() => vec![t.to_string()],
407                _ => vec![],
408            },
409            namespaces: ns_strs,
410            ..Default::default()
411        };
412        let page = self
413            .entities(token)?
414            .query_entities_count_free(
415                token.namespace().as_str(),
416                filter,
417                PageRequest {
418                    offset: offset.into(),
419                    limit,
420                },
421            )
422            .await?;
423        Ok(page.items)
424    }
425
426    /// Count entities filtered by kind and optional domain tag.
427    ///
428    /// Used to report a meaningful `total` alongside a paginated listing (K-6).
429    /// Multi-namespace visibility from the token is applied.
430    pub async fn count_entities_tagged(
431        &self,
432        token: &NamespaceToken,
433        kind: Option<&str>,
434        domain_tag: Option<&str>,
435    ) -> RuntimeResult<u64> {
436        let ns_strs: Vec<String> = token
437            .visible_namespaces()
438            .iter()
439            .map(|ns| ns.as_str().to_owned())
440            .collect();
441        let filter = EntityFilter {
442            kinds: match kind {
443                Some(k) => vec![k.to_string()],
444                None => vec![],
445            },
446            tags_any: match domain_tag {
447                Some(t) if !t.is_empty() => vec![t.to_string()],
448                _ => vec![],
449            },
450            namespaces: ns_strs,
451            ..Default::default()
452        };
453        Ok(self
454            .entities(token)?
455            .count_entities(token.namespace().as_str(), filter)
456            .await?)
457    }
458
459    /// List events in the namespace proven by the caller token.
460    pub async fn list_events(
461        &self,
462        token: &NamespaceToken,
463        filter: EventFilter,
464        page: PageRequest,
465    ) -> RuntimeResult<Page<Event>> {
466        self.events(token)?
467            .query_events(filter, page)
468            .await
469            .map_err(Into::into)
470    }
471
472    /// List notes visible to the token, optionally filtered by kind.
473    ///
474    /// When the token carries a multi-namespace visible set, notes from all
475    /// visible namespaces are returned. When the visible set is `[primary]`
476    /// (the default) this behaves identically to the pre-visibility behaviour.
477    pub async fn list_notes(
478        &self,
479        token: &NamespaceToken,
480        kind: Option<&str>,
481        limit: u32,
482        offset: u32,
483    ) -> RuntimeResult<Vec<Note>> {
484        let visible = token.visible_namespaces();
485        if visible.len() == 1 {
486            // Fast path: single namespace — use the dedicated query_notes method.
487            let page = self
488                .notes(token)?
489                .query_notes_count_free(
490                    token.namespace().as_str(),
491                    kind,
492                    PageRequest {
493                        offset: offset.into(),
494                        limit,
495                    },
496                )
497                .await?;
498            return Ok(page.items);
499        }
500        // Multi-namespace path: use query_notes_filtered with the visible set.
501        use khive_storage::note::NoteFilter;
502        let ns_strs: Vec<String> = visible.iter().map(|ns| ns.as_str().to_owned()).collect();
503        let filter = NoteFilter {
504            kind: kind.map(|k| k.to_string()),
505            namespaces: ns_strs,
506            ..Default::default()
507        };
508        let page = self
509            .notes(token)?
510            .query_notes_filtered_count_free(
511                token.namespace().as_str(),
512                &filter,
513                PageRequest {
514                    offset: offset.into(),
515                    limit,
516                },
517            )
518            .await?;
519        Ok(page.items)
520    }
521
522    /// List an immutable insertion-sequence page of visible notes.
523    ///
524    /// Soft-deleting the prior page's last note does not invalidate the
525    /// cursor because the boundary is resolved including tombstones. A hard
526    /// deletion makes the cursor unresolvable and returns an explicit error.
527    pub async fn list_notes_after(
528        &self,
529        token: &NamespaceToken,
530        kind: Option<&str>,
531        after: Option<Uuid>,
532        limit: u32,
533    ) -> RuntimeResult<(Vec<Note>, Option<Uuid>)> {
534        let store = self.notes(token)?;
535        let after = match after {
536            Some(id) => {
537                let note = self
538                    .get_note_including_deleted(token, id)
539                    .await?
540                    .ok_or_else(|| RuntimeError::NotFound(format!("note cursor {id}")))?;
541                Self::ensure_namespace_visible(&note.namespace, token)?;
542                let sequence = store.note_sequence(id).await?.ok_or_else(|| {
543                    RuntimeError::Internal(format!(
544                        "note cursor {id} has no insertion-sequence ledger row"
545                    ))
546                })?;
547                Some(SeekCursor { sequence, id })
548            }
549            None => None,
550        };
551        let filter = khive_storage::note::NoteFilter {
552            kind: kind.map(str::to_string),
553            namespaces: token
554                .visible_namespaces()
555                .iter()
556                .map(|namespace| namespace.as_str().to_owned())
557                .collect(),
558            ..Default::default()
559        };
560        let page = store
561            .query_notes_filtered_after(token.namespace().as_str(), &filter, after, limit)
562            .await?;
563        Ok((page.items, page.next_after.map(|cursor| cursor.id)))
564    }
565
566    /// Count notes matching `kind` across the caller's visible namespaces.
567    pub async fn count_notes(
568        &self,
569        token: &NamespaceToken,
570        kind: Option<&str>,
571    ) -> RuntimeResult<u64> {
572        let namespaces: Vec<String> = token
573            .visible_namespaces()
574            .iter()
575            .map(|namespace| namespace.as_str().to_owned())
576            .collect();
577        Ok(self
578            .notes(token)?
579            .count_notes_in_namespaces(&namespaces, kind)
580            .await?)
581    }
582
583    /// Count entities in a namespace, optionally filtered.
584    pub async fn count_entities(
585        &self,
586        token: &NamespaceToken,
587        kind: Option<&str>,
588    ) -> RuntimeResult<u64> {
589        let ns_strs: Vec<String> = token
590            .visible_namespaces()
591            .iter()
592            .map(|ns| ns.as_str().to_owned())
593            .collect();
594        let filter = EntityFilter {
595            kinds: match kind {
596                Some(k) => vec![k.to_string()],
597                None => vec![],
598            },
599            namespaces: ns_strs,
600            ..Default::default()
601        };
602        Ok(self
603            .entities(token)?
604            .count_entities(token.namespace().as_str(), filter)
605            .await?)
606    }
607
608    /// Return the coupled entity total and optional type counts for `stats`.
609    pub async fn entity_stats_counts(
610        &self,
611        token: &NamespaceToken,
612    ) -> RuntimeResult<EntityStatsCounts> {
613        entity_stats_counts(self.entities(token)?.as_ref(), token).await
614    }
615
616    /// Validate and prepare one note item for a bulk `create(items=[...])`
617    /// write. The note goes through the preparation a singleton note create
618    /// uses (`validate_head`, then `prepare_atomic_notes`: kind validation,
619    /// owned-identity derivation, secret gate, salience range, row and FTS
620    /// statements) with embedding switched off, so the plan writes the row
621    /// and its FTS document and no vector. A later `reindex` backfills the
622    /// vector, as it does for bulk entities. The caller commits the plan,
623    /// alone or joined with its siblings in one `run_atomic_unit` call.
624    pub async fn prepare_bulk_note_plan(
625        &self,
626        token: &NamespaceToken,
627        spec: NoteCreateSpec,
628    ) -> RuntimeResult<(Note, AtomicOpPlan)> {
629        let mut candidate = Note::new(token.namespace().as_str(), &spec.kind, &spec.content);
630        candidate.name = spec.name.clone();
631        candidate.properties = spec.properties.clone();
632        crate::note_write::validate_head(&candidate)?;
633        let mut prepared = crate::atomic_message::prepare_atomic_notes(
634            self,
635            vec![crate::atomic_message::AtomicNoteSpec {
636                token,
637                id: None,
638                kind: &spec.kind,
639                name: spec.name.as_deref(),
640                content: &spec.content,
641                properties: spec.properties,
642            }],
643            crate::atomic_message::AtomicNoteOptions {
644                salience: spec.salience,
645                embed: Some(false),
646                ..Default::default()
647            },
648        )
649        .await?;
650        match (prepared.notes.pop(), prepared.plans.pop()) {
651            (Some(note), Some(plan)) if prepared.notes.is_empty() && prepared.plans.is_empty() => {
652                Ok((note, plan))
653            }
654            _ => Err(RuntimeError::Internal(
655                "bulk note preparation must yield exactly one note and one plan".into(),
656            )),
657        }
658    }
659}
khive_runtime/operations/record_ops.rs

khive_runtime/operations/
record_ops.rs
