khive_runtime/operations/
entity_bulk.rs

1//! Prepared entity bulk construction.
2
3use super::*;
4use crate::secret_gate_finalizer::entity_admission::{
5    prepare_entity_admission_at, EntityAdmission, EntityEntryPoint,
6};
7use crate::EntityCandidateMutation;
8
9impl KhiveRuntime {
10    /// Create a batch of entities atomically.
11    ///
12    /// All specs are validated before any write. If ANY spec fails validation
13    /// (unknown kind, empty name, secret-gate violation), the method returns
14    /// that error and no entities are written.
15    ///
16    /// Entity rows and their FTS documents are written in one SQLite transaction.
17    /// Any statement failure rolls back the entire batch across both surfaces.
18    /// Embedding is intentionally skipped: bulk structural ingest is the expected
19    /// use-case, and dense vectors are backfilled later via a `reindex` call.
20    pub async fn create_many(
21        &self,
22        token: &NamespaceToken,
23        specs: Vec<EntityCreateSpec>,
24    ) -> RuntimeResult<Vec<Entity>> {
25        if specs.is_empty() {
26            return Ok(vec![]);
27        }
28        let ns = token.namespace().as_str();
29
30        // Phase 1: validate ALL specs before any write.
31        // Includes entity-type validation via the pack-installed validator when available.
32        // Any validation failure here guarantees zero rows are written.
33        let mut admissions = Vec::with_capacity(specs.len());
34        for (index, spec) in specs.iter().enumerate() {
35            let entity = self.validate_bulk_entity(ns, spec)?;
36            admissions.push(prepare_entity_admission_at(
37                token,
38                EntityEntryPoint::Bulk,
39                entity,
40                &format!("entity[{index}]"),
41            )?);
42        }
43
44        #[cfg(any(test, feature = "fault-injection"))]
45        let fts_many_inject = consume_fault(&FTS_FAIL_MANY_NS, ns);
46        #[cfg(not(any(test, feature = "fault-injection")))]
47        let fts_many_inject = false;
48
49        #[cfg(any(test, feature = "fault-injection"))]
50        let fts_many_inject_partial = consume_fault(&FTS_FAIL_MANY_PARTIAL_NS, ns);
51        #[cfg(not(any(test, feature = "fault-injection")))]
52        let fts_many_inject_partial = false;
53
54        let injected_failure_index = if fts_many_inject {
55            Some(0)
56        } else if fts_many_inject_partial {
57            Some(usize::from(admissions.len() > 1))
58        } else {
59            None
60        };
61
62        let _ = self.entities(token)?;
63        let _ = self.text(token)?;
64
65        let mut entities = Vec::with_capacity(admissions.len());
66        let mut plans = Vec::with_capacity(admissions.len());
67        let mut has_finalizer = false;
68        for (index, admission) in admissions.into_iter().enumerate() {
69            has_finalizer |= matches!(&admission, EntityAdmission::Exempt(_));
70            let (entity, plan) = self
71                .finish_bulk_entity_plan(token, admission, injected_failure_index == Some(index))
72                .await?;
73            entities.push(entity);
74            plans.push(plan);
75        }
76
77        match run_atomic_unit(self.sql().as_ref(), plans).await {
78            Ok(AtomicRunOutcome::Committed { .. }) => Ok(entities),
79            Ok(AtomicRunOutcome::RolledBack {
80                failed_op_index,
81                failure,
82            }) => Err(RuntimeError::Internal(format!(
83                "create_many: atomic batch rolled back at entity index {failed_op_index}: \
84                 {failure:?}"
85            ))),
86            Err(e) if has_finalizer => Err(RuntimeError::Storage(e.0)),
87            Err(e) => Err(RuntimeError::Internal(format!(
88                "create_many: atomic batch failed: {}",
89                e.0
90            ))),
91        }
92    }
93
94    /// One bulk entity spec's pre-write checks and its row, shared by
95    /// `create_many` and [`Self::prepare_bulk_entity_plan`] so the two bulk
96    /// entity paths cannot drift apart: kind, entity_type, a nonempty name,
97    /// the reserved secret-gate property and the secret gate itself.
98    fn validate_bulk_entity(&self, ns: &str, spec: &EntityCreateSpec) -> RuntimeResult<Entity> {
99        self.validate_entity_kind(&spec.kind)?;
100        // Validate entity_type at the runtime layer via pack-installed callback.
101        // When no validator is installed (bare runtime, unit tests without packs),
102        // the type passes through unchanged, the same skip-when-absent pattern as
103        // validate_entity_kind. The handler layer remains the primary enforcement point.
104        let validated_type =
105            self.validate_entity_type_for_kind(&spec.kind, spec.entity_type.as_deref())?;
106        if spec.name.trim().is_empty() {
107            return Err(RuntimeError::InvalidInput("name must not be empty".into()));
108        }
109        crate::secret_gate::reject_reserved_secret_gate_property(spec.properties.as_ref())?;
110
111        let mut entity =
112            Entity::new(ns, &spec.kind, &spec.name).with_entity_type(validated_type.as_deref());
113        if let Some(d) = &spec.description {
114            entity = entity.with_description(d);
115        }
116        if let Some(p) = spec.properties.clone() {
117            entity = entity.with_properties(p);
118        }
119        if !spec.tags.is_empty() {
120            entity = entity.with_tags(spec.tags.clone());
121        }
122        Ok(entity)
123    }
124
125    /// Validate and prepare one entity item for a bulk `create(items=[...])`
126    /// write: the same admission and row/FTS plan as [`Self::create_many`],
127    /// with no scheduled reindex, so the vector is deferred to a later
128    /// `reindex` exactly as for `create_many`. The bulk create handler uses
129    /// this for every entity item so entity and note plans can join one
130    /// `run_atomic_unit` call.
131    pub async fn prepare_bulk_entity_plan(
132        &self,
133        token: &NamespaceToken,
134        spec: EntityCreateSpec,
135    ) -> RuntimeResult<(Entity, AtomicOpPlan)> {
136        let entity = self.validate_bulk_entity(token.namespace().as_str(), &spec)?;
137        let admission =
138            prepare_entity_admission_at(token, EntityEntryPoint::Bulk, entity, "entity")?;
139        let _ = self.entities(token)?;
140        let _ = self.text(token)?;
141        self.finish_bulk_entity_plan(token, admission, false).await
142    }
143
144    async fn finish_bulk_entity_plan(
145        &self,
146        token: &NamespaceToken,
147        admission: EntityAdmission,
148        inject_fts_failure: bool,
149    ) -> RuntimeResult<(Entity, AtomicOpPlan)> {
150        match admission {
151            EntityAdmission::Legacy(entity) => {
152                let mut plan = bulk_entity_plan(&entity)?;
153                if inject_fts_failure {
154                    plan.statements.truncate(1);
155                    plan.statements.push(bulk_fts_failure());
156                }
157                Ok((entity, AtomicOpPlan::AddEntity(plan)))
158            }
159            EntityAdmission::Exempt(prepared) => {
160                let entity = prepared.entity().clone();
161                let (mut required, _) = self
162                    .prepare_admitted_entity_indexes(token, &entity, &[], false)
163                    .await?;
164                if inject_fts_failure {
165                    required = vec![bulk_fts_failure()];
166                }
167                let plan = prepared.into_plan(
168                    EntityCandidateMutation::CreateIfAbsent,
169                    required,
170                    PostCommitEffect::None,
171                )?;
172                Ok((entity, AtomicOpPlan::FinalizeEntity(Box::new(plan))))
173            }
174        }
175    }
176}
177
178fn bulk_entity_plan(entity: &Entity) -> RuntimeResult<AddEntityPlan> {
179    crate::secret_gate::reject_reserved_secret_gate_property(entity.properties.as_ref())?;
180    let mut statements = vec![PlanStatement {
181        statement: entity_upsert_statement(entity),
182        guard: Some(AffectedRowGuard::exactly(1)),
183    }];
184    // The FTS insert and rowid-map insert must remain adjacent on one connection.
185    statements.extend(
186        insert_document_statements("fts_entities", &entity_fts_document(entity))
187            .into_iter()
188            .map(|statement| PlanStatement {
189                statement,
190                guard: None,
191            }),
192    );
193    Ok(AddEntityPlan {
194        entity_id: entity.id,
195        statements,
196        post_commit: PostCommitEffect::None,
197    })
198}
199
200fn bulk_fts_failure() -> PlanStatement {
201    PlanStatement {
202        statement: SqlStatement {
203            sql: "INSERT INTO __khive_create_many_injected_failure__ DEFAULT VALUES".into(),
204            params: vec![],
205            label: Some("fts-insert-injected-failure".into()),
206        },
207        guard: None,
208    }
209}
khive_runtime/operations/entity_bulk.rs

khive_runtime/operations/
entity_bulk.rs
