type_bridge/
entity_manager.rs

1#![deny(missing_docs)]
2
3use std::marker::PhantomData;
4use std::sync::Arc;
5
6use type_bridge_contract::id::is_canonical_thing_iid;
7use type_bridge_orm::_manager::DynamicEntityManager;
8use type_bridge_orm::session::backend::TxType;
9use type_bridge_orm::{
10    ProjectedBatchOperation, ProjectedBatchRow, ProjectedCrudExecutor, ProjectedManagerComparison,
11};
12
13use crate::__codegen::{
14    CompleteModel, EncodedCreate, EntityModel, FieldToken, HydrationCapability, IntoEncodedCreate,
15    Model, QueryValued, SubtypeRootModel,
16};
17use crate::entity_codec::{
18    hydrate_entity, lower_entity_create, map_validation_error, resolve_discovered_entity,
19    resolve_entity_authority,
20};
21use crate::error::{Error, ModelValidationPhase};
22use crate::hooks::{CrudOperation, HookRunner, LifecycleHook, ModelKind};
23use crate::projected_batch::{
24    checked_batch_ordinal, create_rows, delete_rows, encode_batch_create, execute_owned_delete,
25    execute_owned_things, prepare_batch, project_encoded_batch_create, reserved_binding_vec,
26    uses_successor_batch_runtime, validate_binding_row_count,
27};
28use crate::projected_codec::{materialize_projected, project_create};
29use crate::projected_filter::ProjectedEntityFilter;
30use crate::schema::Schema;
31use crate::{Database, Result};
32
33#[cfg(test)]
34mod tests;
35
36fn invalid_iid() -> Error {
37    Error::model_validation(
38        ModelValidationPhase::Input,
39        "invalid_iid",
40        vec!["iid".into()],
41        "IID is not canonical",
42        None,
43    )
44}
45
46fn schema_not_bound() -> Error {
47    Error::model_validation(
48        ModelValidationPhase::Input,
49        "schema_not_bound",
50        vec![],
51        "database is not schema-bound",
52        None,
53    )
54}
55
56/// Exact-fetch, hydrate, and materialize one freshly written entity through the
57/// shared open context without any transaction-terminal operation.
58pub(crate) async fn rehydrate_written_entity<M>(
59    manager: &DynamicEntityManager<'_>,
60    iid: &str,
61    id: &type_bridge_contract::id::TypeId,
62    installed: &type_bridge_orm::InstalledRuntimeProjection,
63) -> Result<M>
64where
65    M: crate::__codegen::CompleteModel,
66{
67    let row = manager
68        .get_by_iid_exact(iid)
69        .await
70        .map_err(Error::from_orm)?
71        .ok_or_else(|| {
72            Error::model_validation(
73                ModelValidationPhase::Hydration,
74                "missing_post_write_row",
75                vec!["iid".into()],
76                "written entity was not returned",
77                None,
78            )
79        })?;
80    let hydrated = hydrate_entity(row, id, installed)?;
81    M::materialize(&hydrated, &HydrationCapability::new())
82        .map_err(|error| map_validation_error(error, ModelValidationPhase::Hydration))
83}
84
85impl<S, M> EntitySubtypeManager<'_, S, M>
86where
87    S: Schema,
88    M: SubtypeRootModel<Schema = S> + EntityModel<Schema = S>,
89{
90    /// Reads one canonical IID across the root and its generated concrete descendants.
91    /// Invalid IIDs are rejected before I/O; a valid but absent IID returns `None`.
92    pub async fn get_by_iid(&self, _iid: &str) -> Result<Option<M::Subtypes>> {
93        if !is_canonical_thing_iid(_iid) {
94            return Err(invalid_iid());
95        }
96        let installed = self.db.installed_schema().ok_or_else(schema_not_bound)?;
97        let (_id, descriptor) = resolve_entity_authority(
98            M::TYPE_ID_JSON,
99            installed,
100            ModelValidationPhase::Input,
101            false,
102        )?;
103        let tx = self
104            .db
105            .inner_orm()
106            .transaction_context(TxType::Read)
107            .await
108            .map_err(Error::from_orm)?;
109        let manager =
110            DynamicEntityManager::with_canonical_transaction(tx.clone(), Arc::new(descriptor));
111        let identity = match manager.discover_by_iid(_iid).await {
112            Ok(v) => v,
113            Err(e) => {
114                let _ = tx.close().await;
115                return Err(Error::from_orm(e));
116            }
117        };
118        let out = match identity {
119            None => None,
120            Some(identity) => {
121                let (child_id, child_descriptor) =
122                    match resolve_discovered_entity(&identity.type_name, installed) {
123                        Ok(v) => v,
124                        Err(e) => {
125                            let _ = tx.close().await;
126                            return Err(e);
127                        }
128                    };
129                let child = DynamicEntityManager::with_canonical_transaction(
130                    tx.clone(),
131                    Arc::new(child_descriptor),
132                );
133                let row = match child.get_by_iid_exact(&identity.iid).await {
134                    Ok(Some(v)) => v,
135                    Ok(None) => {
136                        let _ = tx.close().await;
137                        return Err(Error::model_validation(
138                            ModelValidationPhase::Hydration,
139                            "missing_concrete_row",
140                            vec!["iid".into()],
141                            "discovered entity row is missing",
142                            None,
143                        ));
144                    }
145                    Err(e) => {
146                        let _ = tx.close().await;
147                        return Err(Error::from_orm(e));
148                    }
149                };
150                let h = match hydrate_entity(row, &child_id, installed) {
151                    Ok(v) => v,
152                    Err(e) => {
153                        let _ = tx.close().await;
154                        return Err(e);
155                    }
156                };
157                Some(
158                    match M::__tb_dispatch_subtype(&h, &HydrationCapability::new()) {
159                        Ok(v) => v,
160                        Err(e) => {
161                            let _ = tx.close().await;
162                            return Err(map_validation_error(e, ModelValidationPhase::Hydration));
163                        }
164                    },
165                )
166            }
167        };
168        tx.close().await.map_err(Error::from_orm)?;
169        Ok(out)
170    }
171    /// Reads all root/descendant models in application result order, materialized as the
172    /// generated leaf or family result. Validation, database, hydration, and close errors
173    /// are returned without exposing implementation details.
174    pub async fn all(&self) -> Result<Vec<M::Subtypes>> {
175        let installed = self.db.installed_schema().ok_or_else(schema_not_bound)?;
176        let (_id, descriptor) = resolve_entity_authority(
177            M::TYPE_ID_JSON,
178            installed,
179            ModelValidationPhase::Input,
180            false,
181        )?;
182        let tx = self
183            .db
184            .inner_orm()
185            .transaction_context(TxType::Read)
186            .await
187            .map_err(Error::from_orm)?;
188        let manager =
189            DynamicEntityManager::with_canonical_transaction(tx.clone(), Arc::new(descriptor));
190        let identities = match manager.discover_all().await {
191            Ok(v) => v,
192            Err(e) => {
193                let _ = tx.close().await;
194                return Err(Error::from_orm(e));
195            }
196        };
197        let mut out = Vec::with_capacity(identities.len());
198        for identity in identities {
199            let type_json = identity.type_name;
200            let (child_id, child_descriptor) =
201                match resolve_discovered_entity(&type_json, installed) {
202                    Ok(v) => v,
203                    Err(e) => {
204                        let _ = tx.close().await;
205                        return Err(e);
206                    }
207                };
208            let child = DynamicEntityManager::with_canonical_transaction(
209                tx.clone(),
210                Arc::new(child_descriptor),
211            );
212            let row = match child.get_by_iid_exact(&identity.iid).await {
213                Ok(Some(v)) => v,
214                Ok(None) => {
215                    let _ = tx.close().await;
216                    return Err(Error::model_validation(
217                        ModelValidationPhase::Hydration,
218                        "missing_concrete_row",
219                        vec!["iid".into()],
220                        "discovered entity row is missing",
221                        None,
222                    ));
223                }
224                Err(e) => {
225                    let _ = tx.close().await;
226                    return Err(Error::from_orm(e));
227                }
228            };
229            let h = match hydrate_entity(row, &child_id, installed) {
230                Ok(v) => v,
231                Err(e) => {
232                    let _ = tx.close().await;
233                    return Err(e);
234                }
235            };
236            out.push(
237                match M::__tb_dispatch_subtype(&h, &HydrationCapability::new()) {
238                    Ok(v) => v,
239                    Err(e) => {
240                        let _ = tx.close().await;
241                        return Err(map_validation_error(e, ModelValidationPhase::Hydration));
242                    }
243                },
244            );
245        }
246        tx.close().await.map_err(Error::from_orm)?;
247        Ok(out)
248    }
249    /// Counts the root and all concrete descendants using the inclusive subtype scope.
250    pub async fn count(&self) -> Result<u64> {
251        let installed = self.db.installed_schema().ok_or_else(schema_not_bound)?;
252        let (_id, descriptor) = resolve_entity_authority(
253            M::TYPE_ID_JSON,
254            installed,
255            ModelValidationPhase::Input,
256            false,
257        )?;
258        DynamicEntityManager::new_canonical(self.db.inner_orm(), Arc::new(descriptor))
259            .count()
260            .await
261            .map_err(Error::from_orm)
262    }
263}
264
265/// Schema-bound, model-branded manager for exact entity operations.
266/// Exact reads and writes exclude subtypes; methods return client input/schema-validation,
267/// database/transaction/close, or hydration/model-validation errors as applicable.
268pub struct EntityManager<'db, S: Schema, M: EntityModel<Schema = S>> {
269    db: &'db Database<S>,
270    hooks: HookRunner,
271    marker: PhantomData<M>,
272}
273
274/// Read-only, schema/model-branded manager for an inclusive generated subtype association.
275/// Results are the generated associated leaf or closed family type; no writes are exposed.
276/// Reads and counts can return input/schema-validation, database/transaction/close, or
277/// hydration/model-validation errors.
278pub struct EntitySubtypeManager<
279    'db,
280    S: Schema,
281    M: SubtypeRootModel<Schema = S> + EntityModel<Schema = S>,
282> {
283    db: &'db Database<S>,
284    marker: PhantomData<M>,
285}
286
287impl<'db, S: Schema, M: SubtypeRootModel<Schema = S> + EntityModel<Schema = S>>
288    EntitySubtypeManager<'db, S, M>
289{
290    pub(crate) fn new(db: &'db Database<S>) -> Self {
291        Self {
292            db,
293            marker: PhantomData,
294        }
295    }
296}
297
298impl<'db, S, M> EntityManager<'db, S, M>
299where
300    S: Schema,
301    M: SubtypeRootModel<Schema = S> + EntityModel<Schema = S>,
302{
303    /// Switches only the read scope and result shape to the generated inclusive subtype
304    /// association; it does not add write operations.
305    pub fn subtypes(&self) -> EntitySubtypeManager<'db, S, M> {
306        EntitySubtypeManager::new(self.db)
307    }
308}
309
310impl<'db, S: Schema, M: EntityModel<Schema = S>> Clone for EntityManager<'db, S, M> {
311    fn clone(&self) -> Self {
312        Self {
313            db: self.db,
314            hooks: self.hooks.clone(),
315            marker: PhantomData,
316        }
317    }
318}
319
320impl<S: Schema, M: EntityModel<Schema = S>> EntityManager<'_, S, M> {
321    pub(crate) fn new(db: &Database<S>) -> EntityManager<'_, S, M> {
322        EntityManager {
323            db,
324            hooks: HookRunner::default(),
325            marker: PhantomData,
326        }
327    }
328}
329
330impl<'db, S, M> EntityManager<'db, S, M>
331where
332    S: Schema,
333    M: EntityModel<Schema = S> + CompleteModel,
334{
335    /// Start one empty immutable canonical filter for this exact entity model.
336    pub fn filter(&self) -> Result<ProjectedEntityFilter<'db, S, M>> {
337        ProjectedEntityFilter::for_database(self.db)
338    }
339
340    /// Start a canonical filter with one exact generated field comparison.
341    pub fn where_<V>(
342        &self,
343        field: FieldToken<M, V>,
344        operator: ProjectedManagerComparison,
345        value: &V,
346    ) -> Result<ProjectedEntityFilter<'db, S, M>>
347    where
348        V: Model<Schema = S> + QueryValued,
349    {
350        self.filter()?.where_(field, operator, value)
351    }
352
353    fn successor_batches_enabled(&self) -> bool {
354        self.db
355            .installed_schema()
356            .is_some_and(|installed| uses_successor_batch_runtime(installed))
357    }
358
359    /// Register one generated-model lifecycle hook on this manager.
360    /// Pre-hooks run in registration order and post-hooks in reverse order.
361    pub fn add_hook(&mut self, hook: Arc<dyn LifecycleHook>) -> &mut Self {
362        self.hooks.add(hook);
363        self
364    }
365
366    fn encode_hook_input(input: &M::Create) -> Result<EncodedCreate> {
367        input
368            .clone()
369            .into_encoded_create()
370            .map_err(|error| map_validation_error(error, ModelValidationPhase::Input))
371    }
372
373    /// Inserts one exact entity and returns its complete freshly hydrated model. Errors may
374    /// be input/schema validation, database/transaction/close, or model hydration errors.
375    pub async fn insert(&self, input: M::Create) -> Result<M> {
376        if !self.hooks.has_hooks() {
377            return self
378                .projected_write(input, CrudOperation::Insert, None)
379                .await;
380        }
381        let encoded = Self::encode_hook_input(&input)?;
382        let metadata = self
383            .hooks
384            .run_pre(
385                M::TYPE_ID_JSON,
386                ModelKind::Entity,
387                CrudOperation::Insert,
388                None,
389                Some(&encoded),
390            )
391            .await?;
392        let output = self
393            .projected_write(input, CrudOperation::Insert, None)
394            .await?;
395        self.hooks
396            .run_post(
397                M::TYPE_ID_JSON,
398                ModelKind::Entity,
399                CrudOperation::Insert,
400                Some(output.iid()),
401                Some(&encoded),
402                metadata,
403            )
404            .await;
405        Ok(output)
406    }
407    /// Uses a projected exact-model key when one is available; otherwise inserts. For an
408    /// existing exact row, the supplied create value completely replaces non-key ownership:
409    /// omitted optional values and empty multivalue collections remove prior ownership. It
410    /// never reuses a subtype instance, and returns a complete freshly hydrated model.
411    pub async fn put(&self, input: M::Create) -> Result<M> {
412        if !self.hooks.has_hooks() {
413            return self.projected_write(input, CrudOperation::Put, None).await;
414        }
415        let encoded = Self::encode_hook_input(&input)?;
416        let metadata = self
417            .hooks
418            .run_pre(
419                M::TYPE_ID_JSON,
420                ModelKind::Entity,
421                CrudOperation::Put,
422                None,
423                Some(&encoded),
424            )
425            .await?;
426        let output = self
427            .projected_write(input, CrudOperation::Put, None)
428            .await?;
429        self.hooks
430            .run_post(
431                M::TYPE_ID_JSON,
432                ModelKind::Entity,
433                CrudOperation::Put,
434                Some(output.iid()),
435                Some(&encoded),
436                metadata,
437            )
438            .await;
439        Ok(output)
440    }
441    /// Inserts each item and returns complete freshly hydrated models in input order, or one
442    /// error for the whole call with no partial result vector.
443    pub async fn insert_many(&self, inputs: Vec<M::Create>) -> Result<Vec<M>> {
444        let successor = self.successor_batches_enabled();
445        if successor {
446            validate_binding_row_count(inputs.len())?;
447        }
448        if inputs.is_empty() && !successor {
449            return Ok(Vec::new());
450        }
451        if !self.hooks.has_hooks() {
452            return self.write_many(inputs, false).await;
453        }
454        if successor {
455            return self
456                .successor_write_many_with_hooks(
457                    inputs,
458                    CrudOperation::Insert,
459                    ProjectedBatchOperation::Insert,
460                )
461                .await;
462        }
463        let encoded = inputs
464            .iter()
465            .map(Self::encode_hook_input)
466            .collect::<Result<Vec<_>>>()?;
467        let mut metadata = Vec::with_capacity(inputs.len());
468        for input in &encoded {
469            metadata.push(
470                self.hooks
471                    .run_pre(
472                        M::TYPE_ID_JSON,
473                        ModelKind::Entity,
474                        CrudOperation::Insert,
475                        None,
476                        Some(input),
477                    )
478                    .await?,
479            );
480        }
481        let outputs = self.write_many(inputs.clone(), false).await?;
482        for ((input, output), metadata) in encoded.iter().zip(&outputs).zip(metadata) {
483            self.hooks
484                .run_post(
485                    M::TYPE_ID_JSON,
486                    ModelKind::Entity,
487                    CrudOperation::Insert,
488                    Some(output.iid()),
489                    Some(input),
490                    metadata,
491                )
492                .await;
493        }
494        Ok(outputs)
495    }
496    /// Applies the per-item [`Self::put`] key-or-insert rule, including complete replacement of
497    /// non-key ownership for existing exact rows, returning complete freshly hydrated models
498    /// in input order, or one error for the whole call with no partial vector.
499    pub async fn put_many(&self, inputs: Vec<M::Create>) -> Result<Vec<M>> {
500        let successor = self.successor_batches_enabled();
501        if successor {
502            validate_binding_row_count(inputs.len())?;
503        }
504        if inputs.is_empty() && !successor {
505            return Ok(Vec::new());
506        }
507        if !self.hooks.has_hooks() {
508            return self.write_many(inputs, true).await;
509        }
510        if successor {
511            return self
512                .successor_write_many_with_hooks(
513                    inputs,
514                    CrudOperation::Put,
515                    ProjectedBatchOperation::Put,
516                )
517                .await;
518        }
519        let encoded = inputs
520            .iter()
521            .map(Self::encode_hook_input)
522            .collect::<Result<Vec<_>>>()?;
523        let mut metadata = Vec::with_capacity(inputs.len());
524        for input in &encoded {
525            metadata.push(
526                self.hooks
527                    .run_pre(
528                        M::TYPE_ID_JSON,
529                        ModelKind::Entity,
530                        CrudOperation::Put,
531                        None,
532                        Some(input),
533                    )
534                    .await?,
535            );
536        }
537        let outputs = self.write_many(inputs.clone(), true).await?;
538        for ((input, output), metadata) in encoded.iter().zip(&outputs).zip(metadata) {
539            self.hooks
540                .run_post(
541                    M::TYPE_ID_JSON,
542                    ModelKind::Entity,
543                    CrudOperation::Put,
544                    Some(output.iid()),
545                    Some(input),
546                    metadata,
547                )
548                .await;
549        }
550        Ok(outputs)
551    }
552    /// Completely replaces non-key ownership on the exact model at canonical `iid`, preserves
553    /// that IID, and returns its complete freshly hydrated model. Omitted optional values and
554    /// empty multivalue collections remove prior ownership.
555    pub async fn update(&self, iid: &str, input: M::Create) -> Result<M> {
556        if !is_canonical_thing_iid(iid) {
557            return Err(invalid_iid());
558        }
559        if !self.hooks.has_hooks() {
560            return self
561                .projected_write(input, CrudOperation::Update, Some(iid))
562                .await;
563        }
564        let encoded = Self::encode_hook_input(&input)?;
565        let state = self
566            .hooks
567            .run_pre(
568                M::TYPE_ID_JSON,
569                ModelKind::Entity,
570                CrudOperation::Update,
571                Some(iid),
572                Some(&encoded),
573            )
574            .await?;
575        let output = self
576            .projected_write(input, CrudOperation::Update, Some(iid))
577            .await?;
578        self.hooks
579            .run_post(
580                M::TYPE_ID_JSON,
581                ModelKind::Entity,
582                CrudOperation::Update,
583                Some(output.iid()),
584                Some(&encoded),
585                state,
586            )
587            .await;
588        Ok(output)
589    }
590
591    /// Deletes only the exact model at canonical `iid`; subtype instances are not targeted.
592    pub async fn delete(&self, iid: &str) -> Result<()> {
593        if !is_canonical_thing_iid(iid) {
594            return Err(invalid_iid());
595        }
596        if !self.hooks.has_hooks() {
597            return self.projected_delete(iid).await;
598        }
599        let state = self
600            .hooks
601            .run_pre(
602                M::TYPE_ID_JSON,
603                ModelKind::Entity,
604                CrudOperation::Delete,
605                Some(iid),
606                None,
607            )
608            .await?;
609        self.projected_delete(iid).await?;
610        self.hooks
611            .run_post(
612                M::TYPE_ID_JSON,
613                ModelKind::Entity,
614                CrudOperation::Delete,
615                Some(iid),
616                None,
617                state,
618            )
619            .await;
620        Ok(())
621    }
622
623    /// Atomically replaces each exact entity identified by its canonical IID and returns
624    /// complete freshly hydrated models in input order. Every input and pre-hook completes
625    /// before database work begins; any write or hydration failure rolls back the whole batch.
626    pub async fn update_many(&self, inputs: Vec<(String, M::Create)>) -> Result<Vec<M>> {
627        let successor = self.successor_batches_enabled();
628        if successor {
629            validate_binding_row_count(inputs.len())?;
630        }
631        if inputs.is_empty() && !successor {
632            return Ok(Vec::new());
633        }
634        let installed = self.db.installed_schema().ok_or_else(schema_not_bound)?;
635        let (id, descriptor) = resolve_entity_authority(
636            M::TYPE_ID_JSON,
637            installed,
638            ModelValidationPhase::Input,
639            true,
640        )?;
641        if uses_successor_batch_runtime(installed) {
642            if !self.hooks.has_hooks() {
643                let rows = crate::projected_batch::update_rows(installed, &id, inputs)?;
644                let batch = prepare_batch(installed, id, ProjectedBatchOperation::Update, rows)?;
645                return execute_owned_things(self.db.inner_orm(), installed, &batch).await;
646            }
647            let mut hook_inputs = reserved_binding_vec(inputs.len())?;
648            let mut rows = reserved_binding_vec(inputs.len())?;
649            for (ordinal, (iid, input)) in inputs.into_iter().enumerate() {
650                let encoded = encode_batch_create(input, ordinal)?;
651                rows.push(ProjectedBatchRow::Update {
652                    iid: iid.clone(),
653                    replacement: project_encoded_batch_create(&encoded, &id, installed, ordinal)?,
654                });
655                hook_inputs.push((iid, encoded));
656            }
657            let batch = prepare_batch(installed, id, ProjectedBatchOperation::Update, rows)?;
658
659            let mut states = reserved_binding_vec(hook_inputs.len())?;
660            for (ordinal, (iid, encoded)) in hook_inputs.iter().enumerate() {
661                states.push(
662                    self.hooks
663                        .run_pre(
664                            M::TYPE_ID_JSON,
665                            ModelKind::Entity,
666                            CrudOperation::Update,
667                            Some(iid),
668                            Some(encoded),
669                        )
670                        .await
671                        .map_err(|error| {
672                            error.with_projected_batch_row(checked_batch_ordinal(ordinal))
673                        })?,
674                );
675            }
676
677            let outputs = execute_owned_things::<M>(self.db.inner_orm(), installed, &batch).await?;
678            for (((_, encoded), output), state) in hook_inputs.iter().zip(&outputs).zip(states) {
679                self.hooks
680                    .run_post(
681                        M::TYPE_ID_JSON,
682                        ModelKind::Entity,
683                        CrudOperation::Update,
684                        Some(output.iid()),
685                        Some(encoded),
686                        state,
687                    )
688                    .await;
689            }
690            return Ok(outputs);
691        }
692        let mut prepared = Vec::with_capacity(inputs.len());
693        for (iid, input) in inputs {
694            if !is_canonical_thing_iid(&iid) {
695                return Err(invalid_iid());
696            }
697            let encoded = Self::encode_hook_input(&input)?;
698            let attributes = lower_entity_create(input, &id, installed)?;
699            prepared.push((iid, encoded, attributes));
700        }
701
702        let mut states = Vec::new();
703        if self.hooks.has_hooks() {
704            states.reserve(prepared.len());
705            for (iid, encoded, _) in &prepared {
706                states.push(
707                    self.hooks
708                        .run_pre(
709                            M::TYPE_ID_JSON,
710                            ModelKind::Entity,
711                            CrudOperation::Update,
712                            Some(iid),
713                            Some(encoded),
714                        )
715                        .await?,
716                );
717            }
718        }
719
720        let tx = self
721            .db
722            .inner_orm()
723            .transaction_context(TxType::Write)
724            .await
725            .map_err(Error::from_orm)?;
726        let manager =
727            DynamicEntityManager::with_canonical_transaction(tx.clone(), Arc::new(descriptor));
728        let mut outputs = Vec::with_capacity(prepared.len());
729        for (iid, _, attributes) in &prepared {
730            if let Err(error) = manager.update_exact(iid, attributes).await {
731                let _ = tx.rollback().await;
732                return Err(Error::from_orm(error));
733            }
734            match rehydrate_written_entity::<M>(&manager, iid, &id, installed).await {
735                Ok(output) => outputs.push(output),
736                Err(error) => {
737                    let _ = tx.rollback().await;
738                    return Err(error);
739                }
740            }
741        }
742        tx.commit().await.map_err(Error::from_orm)?;
743
744        if self.hooks.has_hooks() {
745            for (((_, encoded, _), output), state) in prepared.iter().zip(&outputs).zip(states) {
746                self.hooks
747                    .run_post(
748                        M::TYPE_ID_JSON,
749                        ModelKind::Entity,
750                        CrudOperation::Update,
751                        Some(output.iid()),
752                        Some(encoded),
753                        state,
754                    )
755                    .await;
756            }
757        }
758        Ok(outputs)
759    }
760
761    /// Atomically deletes every exact entity at the supplied canonical IIDs. All IIDs and
762    /// pre-hooks are accepted before database work begins; any write failure rolls back the
763    /// whole batch, and post-hook failures do not change the committed result.
764    pub async fn delete_many(&self, iids: &[String]) -> Result<()> {
765        let successor = self.successor_batches_enabled();
766        if successor {
767            validate_binding_row_count(iids.len())?;
768        }
769        if iids.is_empty() && !successor {
770            return Ok(());
771        }
772        if !successor && iids.iter().any(|iid| !is_canonical_thing_iid(iid)) {
773            return Err(invalid_iid());
774        }
775        let installed = self.db.installed_schema().ok_or_else(schema_not_bound)?;
776        let (id, descriptor) = resolve_entity_authority(
777            M::TYPE_ID_JSON,
778            installed,
779            ModelValidationPhase::Input,
780            true,
781        )?;
782
783        let batch = if successor {
784            Some(prepare_batch(
785                installed,
786                id.clone(),
787                ProjectedBatchOperation::Delete,
788                delete_rows(iids)?,
789            )?)
790        } else {
791            None
792        };
793        let mut states = if successor && self.hooks.has_hooks() {
794            reserved_binding_vec(iids.len())?
795        } else {
796            Vec::new()
797        };
798        if self.hooks.has_hooks() {
799            if !successor {
800                states.reserve(iids.len());
801            }
802            for (ordinal, iid) in iids.iter().enumerate() {
803                states.push(
804                    self.hooks
805                        .run_pre(
806                            M::TYPE_ID_JSON,
807                            ModelKind::Entity,
808                            CrudOperation::Delete,
809                            Some(iid),
810                            None,
811                        )
812                        .await
813                        .map_err(|error| {
814                            if successor {
815                                error.with_projected_batch_row(checked_batch_ordinal(ordinal))
816                            } else {
817                                error
818                            }
819                        })?,
820                );
821            }
822        }
823
824        if uses_successor_batch_runtime(installed) {
825            execute_owned_delete(
826                self.db.inner_orm(),
827                installed,
828                batch.as_ref().expect("successor batch was prepared"),
829            )
830            .await?;
831        } else {
832            let tx = self
833                .db
834                .inner_orm()
835                .transaction_context(TxType::Write)
836                .await
837                .map_err(Error::from_orm)?;
838            let manager =
839                DynamicEntityManager::with_canonical_transaction(tx.clone(), Arc::new(descriptor));
840            for iid in iids {
841                if let Err(error) = manager.delete_by_iid_exact(iid).await {
842                    let _ = tx.rollback().await;
843                    return Err(Error::from_orm(error));
844                }
845            }
846            tx.commit().await.map_err(Error::from_orm)?;
847        }
848
849        if self.hooks.has_hooks() {
850            for (iid, state) in iids.iter().zip(states) {
851                self.hooks
852                    .run_post(
853                        M::TYPE_ID_JSON,
854                        ModelKind::Entity,
855                        CrudOperation::Delete,
856                        Some(iid),
857                        None,
858                        state,
859                    )
860                    .await;
861            }
862        }
863        Ok(())
864    }
865
866    async fn projected_write(
867        &self,
868        input: M::Create,
869        operation: CrudOperation,
870        iid: Option<&str>,
871    ) -> Result<M> {
872        let installed = self.db.installed_schema().ok_or_else(schema_not_bound)?;
873        let (id, _descriptor) = resolve_entity_authority(
874            M::TYPE_ID_JSON,
875            installed,
876            ModelValidationPhase::Input,
877            true,
878        )?;
879        let create = project_create(input, &id, installed)?;
880        let tx = self
881            .db
882            .inner_orm()
883            .transaction_context(TxType::Write)
884            .await
885            .map_err(Error::from_orm)?;
886        let executor = ProjectedCrudExecutor::new(installed);
887        let projected = match operation {
888            CrudOperation::Insert => {
889                executor
890                    .insert_entity_in_transaction_with_compatibility(&tx, &create)
891                    .await
892            }
893            CrudOperation::Put => {
894                executor
895                    .put_entity_in_transaction_with_compatibility(&tx, &create)
896                    .await
897            }
898            CrudOperation::Update => {
899                executor
900                    .update_entity_in_transaction_with_compatibility(
901                        &tx,
902                        iid.expect("projected entity update requires a checked IID"),
903                        &create,
904                    )
905                    .await
906            }
907            CrudOperation::Delete => unreachable!("delete has no generated create payload"),
908        };
909        let projected = match projected {
910            Ok(value) => value,
911            Err(error) => {
912                let mapped = Error::from_projected_crud(error, ModelKind::Entity, Some(operation));
913                let _ = tx.rollback().await;
914                return Err(mapped);
915            }
916        };
917        let value = match materialize_projected(projected, installed) {
918            Ok(value) => value,
919            Err(error) => {
920                let _ = tx.rollback().await;
921                return Err(error);
922            }
923        };
924        tx.commit_classified()
925            .await
926            .map_err(|error| Error::from_orm(error.into_orm_error()))?;
927        Ok(value)
928    }
929
930    async fn projected_delete(&self, iid: &str) -> Result<()> {
931        let installed = self.db.installed_schema().ok_or_else(schema_not_bound)?;
932        let (id, _descriptor) = resolve_entity_authority(
933            M::TYPE_ID_JSON,
934            installed,
935            ModelValidationPhase::Input,
936            true,
937        )?;
938        let tx = self
939            .db
940            .inner_orm()
941            .transaction_context(TxType::Write)
942            .await
943            .map_err(Error::from_orm)?;
944        let result = ProjectedCrudExecutor::new(installed)
945            .delete_entity_by_iid_in_transaction_with_compatibility(&tx, &id, iid)
946            .await;
947        if let Err(error) = result {
948            let mapped =
949                Error::from_projected_crud(error, ModelKind::Entity, Some(CrudOperation::Delete));
950            let _ = tx.rollback().await;
951            return Err(mapped);
952        }
953        tx.commit_classified()
954            .await
955            .map_err(|error| Error::from_orm(error.into_orm_error()))
956    }
957
958    async fn successor_write_many_with_hooks(
959        &self,
960        inputs: Vec<M::Create>,
961        crud_operation: CrudOperation,
962        batch_operation: ProjectedBatchOperation,
963    ) -> Result<Vec<M>> {
964        let installed = self.db.installed_schema().ok_or_else(schema_not_bound)?;
965        let (id, _descriptor) = resolve_entity_authority(
966            M::TYPE_ID_JSON,
967            installed,
968            ModelValidationPhase::Input,
969            true,
970        )?;
971        let mut encoded = reserved_binding_vec(inputs.len())?;
972        let mut rows = reserved_binding_vec(inputs.len())?;
973        for (ordinal, input) in inputs.into_iter().enumerate() {
974            let input = encode_batch_create(input, ordinal)?;
975            rows.push(ProjectedBatchRow::Create(project_encoded_batch_create(
976                &input, &id, installed, ordinal,
977            )?));
978            encoded.push(input);
979        }
980        let batch = prepare_batch(installed, id, batch_operation, rows)?;
981
982        let mut metadata = reserved_binding_vec(encoded.len())?;
983        for (ordinal, input) in encoded.iter().enumerate() {
984            metadata.push(
985                self.hooks
986                    .run_pre(
987                        M::TYPE_ID_JSON,
988                        ModelKind::Entity,
989                        crud_operation,
990                        None,
991                        Some(input),
992                    )
993                    .await
994                    .map_err(|error| {
995                        error.with_projected_batch_row(checked_batch_ordinal(ordinal))
996                    })?,
997            );
998        }
999
1000        let outputs = execute_owned_things::<M>(self.db.inner_orm(), installed, &batch).await?;
1001        for ((input, output), metadata) in encoded.iter().zip(&outputs).zip(metadata) {
1002            self.hooks
1003                .run_post(
1004                    M::TYPE_ID_JSON,
1005                    ModelKind::Entity,
1006                    crud_operation,
1007                    Some(output.iid()),
1008                    Some(input),
1009                    metadata,
1010                )
1011                .await;
1012        }
1013        Ok(outputs)
1014    }
1015
1016    async fn write_many(&self, inputs: Vec<M::Create>, put: bool) -> Result<Vec<M>> {
1017        let installed = self.db.installed_schema().ok_or_else(schema_not_bound)?;
1018        let (id, descriptor) = resolve_entity_authority(
1019            M::TYPE_ID_JSON,
1020            installed,
1021            ModelValidationPhase::Input,
1022            true,
1023        )?;
1024        if uses_successor_batch_runtime(installed) {
1025            let rows = create_rows(installed, &id, inputs)?;
1026            let operation = if put {
1027                ProjectedBatchOperation::Put
1028            } else {
1029                ProjectedBatchOperation::Insert
1030            };
1031            let batch = prepare_batch(installed, id, operation, rows)?;
1032            return execute_owned_things(self.db.inner_orm(), installed, &batch).await;
1033        }
1034        let mut lowered = Vec::with_capacity(inputs.len());
1035        for input in inputs {
1036            lowered.push(lower_entity_create(input, &id, installed)?);
1037        }
1038        let tx = self
1039            .db
1040            .inner_orm()
1041            .transaction_context(TxType::Write)
1042            .await
1043            .map_err(Error::from_orm)?;
1044        let manager =
1045            DynamicEntityManager::with_canonical_transaction(tx.clone(), Arc::new(descriptor));
1046        let iids = match if put {
1047            manager.put_many_exact(&lowered).await
1048        } else {
1049            manager.insert_many(&lowered).await
1050        } {
1051            Ok(v) if v.len() == lowered.len() => v,
1052            Ok(_) => {
1053                let _ = tx.rollback().await;
1054                return Err(Error::model_validation(
1055                    ModelValidationPhase::Hydration,
1056                    "iid_count_mismatch",
1057                    vec!["iid".into()],
1058                    "provider returned an unexpected IID count",
1059                    None,
1060                ));
1061            }
1062            Err(e) => {
1063                let _ = tx.rollback().await;
1064                return Err(Error::from_orm(e));
1065            }
1066        };
1067        let mut out = Vec::with_capacity(iids.len());
1068        for iid in iids {
1069            let row = match manager.get_by_iid_exact(&iid).await {
1070                Ok(Some(r)) => r,
1071                Ok(None) => {
1072                    let _ = tx.rollback().await;
1073                    return Err(Error::model_validation(
1074                        ModelValidationPhase::Hydration,
1075                        "missing_post_write_row",
1076                        vec!["iid".into()],
1077                        "written entity was not returned",
1078                        None,
1079                    ));
1080                }
1081                Err(e) => {
1082                    let _ = tx.rollback().await;
1083                    return Err(Error::from_orm(e));
1084                }
1085            };
1086            let h = match hydrate_entity(row, &id, installed) {
1087                Ok(v) => v,
1088                Err(e) => {
1089                    let _ = tx.rollback().await;
1090                    return Err(e);
1091                }
1092            };
1093            let value = match M::materialize(&h, &HydrationCapability::new()) {
1094                Ok(v) => v,
1095                Err(e) => {
1096                    let _ = tx.rollback().await;
1097                    return Err(map_validation_error(e, ModelValidationPhase::Hydration));
1098                }
1099            };
1100            out.push(value);
1101        }
1102        tx.commit().await.map_err(Error::from_orm)?;
1103        Ok(out)
1104    }
1105    /// Counts only exact models, excluding subtypes.
1106    pub async fn count(&self) -> Result<u64> {
1107        let installed = self.db.installed_schema().ok_or_else(schema_not_bound)?;
1108        let (id, _descriptor) = resolve_entity_authority(
1109            M::TYPE_ID_JSON,
1110            installed,
1111            ModelValidationPhase::Input,
1112            true,
1113        )?;
1114        ProjectedCrudExecutor::new(installed)
1115            .count_entities_with_compatibility(self.db.inner_orm(), &id)
1116            .await
1117            .map_err(|error| Error::from_projected_crud(error, ModelKind::Entity, None))
1118    }
1119
1120    /// Reads one exact model by canonical IID; invalid IIDs are rejected before I/O and a
1121    /// valid but absent model returns `None`.
1122    pub async fn get_by_iid(&self, iid: &str) -> Result<Option<M>> {
1123        if !is_canonical_thing_iid(iid) {
1124            return Err(invalid_iid());
1125        }
1126        let installed = self.db.installed_schema().ok_or_else(schema_not_bound)?;
1127        let (id, _descriptor) = resolve_entity_authority(
1128            M::TYPE_ID_JSON,
1129            installed,
1130            ModelValidationPhase::Input,
1131            true,
1132        )?;
1133        ProjectedCrudExecutor::new(installed)
1134            .get_entity_by_iid_with_compatibility(self.db.inner_orm(), &id, iid)
1135            .await
1136            .map_err(|error| Error::from_projected_crud(error, ModelKind::Entity, None))?
1137            .map(|projected| materialize_projected(projected, installed))
1138            .transpose()
1139    }
1140
1141    /// Reads all exact models in application result order, excluding subtypes; each result is
1142    /// a complete hydrated model.
1143    pub async fn all(&self) -> Result<Vec<M>> {
1144        let installed = self.db.installed_schema().ok_or_else(schema_not_bound)?;
1145        let (id, descriptor) = resolve_entity_authority(
1146            M::TYPE_ID_JSON,
1147            installed,
1148            ModelValidationPhase::Input,
1149            true,
1150        )?;
1151        let rows =
1152            DynamicEntityManager::new_canonical(self.db.inner_orm(), Arc::new(descriptor.clone()))
1153                .all_exact()
1154                .await
1155                .map_err(Error::from_orm)?;
1156        rows.into_iter()
1157            .map(|r| {
1158                let h = hydrate_entity(r, &id, installed)?;
1159                M::materialize(&h, &HydrationCapability::new())
1160                    .map_err(|e| map_validation_error(e, ModelValidationPhase::Hydration))
1161            })
1162            .collect()
1163    }
1164}
type_bridge/entity_manager.rs

type_bridge/
entity_manager.rs
