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