1use std::any::{Any, TypeId};
2use std::borrow::Cow;
3use std::collections::{BTreeMap, BTreeSet, HashMap};
4use std::sync::{Arc, Mutex, OnceLock, Weak};
5
6use teaql_core::{EntitySnapshot, MutationValues, SmartList, Value};
7
8#[derive(Debug, Clone, Copy, PartialEq, Eq)]
14pub enum LoadedRelation {
15 Loaded,
16 Empty,
17 NotLoaded,
18}
19
20#[derive(Debug, Clone, Copy)]
25pub struct RelationHandle<'a, T> {
26 state: LoadedRelation,
27 value: Option<&'a T>,
28}
29
30impl<'a, T> RelationHandle<'a, T> {
31 fn new(state: LoadedRelation, value: Option<&'a T>) -> Self {
32 Self { state, value }
33 }
34
35 pub fn state(&self) -> LoadedRelation {
36 self.state
37 }
38
39 pub fn value(&self) -> Option<&'a T> {
40 self.value
41 }
42
43 pub fn is_loaded(&self) -> bool {
44 self.state != LoadedRelation::NotLoaded
45 }
46
47 pub fn is_empty(&self) -> bool {
48 self.state == LoadedRelation::Empty
49 }
50}
51
52#[derive(Debug, Clone)]
53pub struct EntityKey {
54 pub entity: Cow<'static, str>,
55 pub id: Value,
56 id_key: EntityIdentityKey,
57}
58
59#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord)]
60enum EntityIdentityKey {
61 Null,
62 Bool(bool),
63 I64(i64),
64 U64(u64),
65 F64(u64),
66 Decimal(rust_decimal::Decimal),
67 Text(String),
68 Date(chrono::NaiveDate),
69 Timestamp(i64),
70 Other(String),
71}
72
73impl EntityKey {
74 pub fn new(entity: impl Into<String>, id: impl Into<Value>) -> Self {
75 let id = id.into();
76 Self {
77 entity: Cow::Owned(entity.into()),
78 id_key: entity_identity_key(&id),
79 id,
80 }
81 }
82
83 pub fn new_static(entity: &'static str, id: impl Into<Value>) -> Self {
84 let id = id.into();
85 Self {
86 entity: Cow::Borrowed(entity),
87 id_key: entity_identity_key(&id),
88 id,
89 }
90 }
91}
92
93impl PartialEq for EntityKey {
94 fn eq(&self, other: &Self) -> bool {
95 self.entity == other.entity && self.id_key == other.id_key
96 }
97}
98
99impl Eq for EntityKey {}
100
101impl PartialOrd for EntityKey {
102 fn partial_cmp(&self, other: &Self) -> Option<std::cmp::Ordering> {
103 Some(self.cmp(other))
104 }
105}
106
107impl Ord for EntityKey {
108 fn cmp(&self, other: &Self) -> std::cmp::Ordering {
109 self.entity
110 .cmp(&other.entity)
111 .then_with(|| self.id_key.cmp(&other.id_key))
112 }
113}
114
115fn entity_identity_key(value: &Value) -> EntityIdentityKey {
116 match value {
117 Value::Null | Value::TypedNull(_) => EntityIdentityKey::Null,
118 Value::Bool(value) => EntityIdentityKey::Bool(*value),
119 Value::I64(value) if *value >= 0 => EntityIdentityKey::U64(*value as u64),
120 Value::I64(value) => EntityIdentityKey::I64(*value),
121 Value::U64(value) => EntityIdentityKey::U64(*value),
122 Value::F64(value) => EntityIdentityKey::F64(value.to_bits()),
123 Value::Decimal(value) => EntityIdentityKey::Decimal(*value),
124 Value::Text(value) => EntityIdentityKey::Text(value.clone()),
125 Value::Json(value) => EntityIdentityKey::Other(format!("json:{value}")),
126 Value::Date(value) => EntityIdentityKey::Date(*value),
127 Value::Timestamp(value) => EntityIdentityKey::Timestamp(value.0),
128 Value::Object(_) => EntityIdentityKey::Other("object".to_owned()),
129 Value::List(_) => EntityIdentityKey::Other("list".to_owned()),
130 }
131}
132
133#[derive(Default)]
134pub struct EntityGraphBuilder {
135 tables: HashMap<TypeId, EntityTable>,
136 relation_lists: HashMap<RelationListKey, Box<dyn Any + Send + Sync>>,
137}
138
139type EntityTable = HashMap<u64, Box<dyn Any + Send + Sync>>;
140
141#[derive(Debug, Clone, PartialEq, Eq, Hash)]
142struct RelationListKey {
143 owner_entity: String,
144 owner_id: u64,
145 relation: String,
146}
147
148impl EntityGraphBuilder {
149 pub fn install<T>(&mut self, id: u64, entity: T)
150 where
151 T: Any + Send + Sync,
152 {
153 self.tables
154 .entry(TypeId::of::<T>())
155 .or_default()
156 .insert(id, Box::new(entity));
157 }
158
159 pub fn entity_count(&self) -> usize {
160 self.tables.values().map(HashMap::len).sum()
161 }
162
163 pub fn install_relation_list<T>(
164 &mut self,
165 owner_entity: impl Into<String>,
166 owner_id: u64,
167 relation: impl Into<String>,
168 list: SmartList<T>,
169 ) where
170 T: Any + Send + Sync,
171 {
172 self.relation_lists.insert(
173 RelationListKey {
174 owner_entity: crate::canonical_id_space_entity(&owner_entity.into()),
175 owner_id,
176 relation: relation.into(),
177 },
178 Box::new(list),
179 );
180 }
181
182 pub fn install_relation_option<T>(
183 &mut self,
184 owner_entity: impl Into<String>,
185 owner_id: u64,
186 relation: impl Into<String>,
187 value: Option<T>,
188 ) where
189 T: Any + Send + Sync,
190 {
191 self.relation_lists.insert(
192 RelationListKey {
193 owner_entity: crate::canonical_id_space_entity(&owner_entity.into()),
194 owner_id,
195 relation: relation.into(),
196 },
197 Box::new(value),
198 );
199 }
200
201 pub fn relation_list_count(&self) -> usize {
202 self.relation_lists.len()
203 }
204
205 fn freeze(self) -> FrozenEntityGraph {
206 FrozenEntityGraph {
207 tables: self.tables,
208 relation_lists: self.relation_lists,
209 }
210 }
211}
212
213impl std::fmt::Debug for EntityGraphBuilder {
214 fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
215 formatter
216 .debug_struct("EntityGraphBuilder")
217 .field("entity_types", &self.tables.len())
218 .field("entities", &self.entity_count())
219 .field("relation_lists", &self.relation_list_count())
220 .finish()
221 }
222}
223
224struct FrozenEntityGraph {
225 tables: HashMap<TypeId, EntityTable>,
226 relation_lists: HashMap<RelationListKey, Box<dyn Any + Send + Sync>>,
227}
228
229impl std::fmt::Debug for FrozenEntityGraph {
230 fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
231 formatter
232 .debug_struct("FrozenEntityGraph")
233 .field("entity_types", &self.tables.len())
234 .field(
235 "entities",
236 &self.tables.values().map(HashMap::len).sum::<usize>(),
237 )
238 .field("relation_lists", &self.relation_lists.len())
239 .finish()
240 }
241}
242
243#[derive(Debug, Clone, Default, PartialEq)]
244pub struct EntityChangeSet {
245 changes: BTreeMap<EntityKey, MutationValues>,
246}
247
248#[derive(Debug, Clone, Default)]
249struct OriginalVersions {
250 first: Option<(EntityKey, i64)>,
251 overflow: BTreeMap<EntityKey, i64>,
252}
253
254impl OriginalVersions {
255 fn clear(&mut self) {
256 self.first = None;
257 self.overflow.clear();
258 }
259
260 fn get(&self, key: &EntityKey) -> Option<i64> {
261 self.first
262 .as_ref()
263 .and_then(|(first_key, version)| (first_key == key).then_some(*version))
264 .or_else(|| self.overflow.get(key).copied())
265 }
266
267 fn insert(&mut self, key: EntityKey, version: i64) {
268 match &mut self.first {
269 None => self.first = Some((key, version)),
270 Some((first_key, first_version)) if first_key == &key => *first_version = version,
271 Some(_) => {
272 self.overflow.insert(key, version);
273 }
274 }
275 }
276
277 fn merge_from(&mut self, source: &Self) {
278 if let Some((key, version)) = &source.first {
279 self.insert(key.clone(), *version);
280 }
281 for (key, version) in &source.overflow {
282 self.insert(key.clone(), *version);
283 }
284 }
285}
286
287impl EntityChangeSet {
288 pub fn is_empty(&self) -> bool {
289 self.changes.is_empty()
290 }
291
292 pub fn set(&mut self, key: EntityKey, field: impl Into<String>, value: Value) {
293 self.changes
294 .entry(key)
295 .or_default()
296 .insert(field.into(), value);
297 }
298
299 pub fn get(&self, key: &EntityKey, field: &str) -> Option<&Value> {
300 self.changes.get(key).and_then(|changes| changes.get(field))
301 }
302
303 pub fn changes(&self) -> &BTreeMap<EntityKey, MutationValues> {
304 &self.changes
305 }
306
307 pub fn clear_entity(&mut self, key: &EntityKey) {
309 self.changes.remove(key);
310 }
311
312 pub fn field_names(&self, key: &EntityKey) -> BTreeSet<String> {
314 self.changes
315 .get(key)
316 .map(|record| record.keys().cloned().collect())
317 .unwrap_or_default()
318 }
319}
320
321#[derive(Debug, Clone, Default, PartialEq)]
322pub struct ChangeSetStack {
323 stack: Vec<EntityChangeSet>,
324}
325
326impl ChangeSetStack {
327 pub fn current_mut(&mut self) -> &mut EntityChangeSet {
328 if self.stack.is_empty() {
329 self.stack.push(EntityChangeSet::default());
330 }
331 self.stack.last_mut().expect("change set stack has current")
332 }
333
334 pub fn current(&self) -> Option<&EntityChangeSet> {
335 self.stack.last()
336 }
337
338 pub fn push(&mut self) {
339 self.stack.push(EntityChangeSet::default());
340 }
341
342 pub fn pop(&mut self) -> Option<EntityChangeSet> {
343 self.stack.pop()
344 }
345
346 pub fn get(&self, key: &EntityKey, field: &str) -> Option<Value> {
347 self.stack
348 .iter()
349 .rev()
350 .find_map(|change_set| change_set.get(key, field).cloned())
351 }
352
353 pub fn set(&mut self, key: EntityKey, field: impl Into<String>, value: Value) {
354 self.current_mut().set(key, field, value);
355 }
356
357 pub fn clear_current(&mut self) {
358 if let Some(current) = self.stack.last_mut() {
359 *current = EntityChangeSet::default();
360 }
361 }
362
363 pub fn clear_entity(&mut self, key: &EntityKey) {
365 for change_set in &mut self.stack {
366 change_set.clear_entity(key);
367 }
368 }
369
370 pub fn changed_field_names(&self, key: &EntityKey) -> BTreeSet<String> {
373 let mut fields = BTreeSet::new();
374 for change_set in &self.stack {
375 fields.extend(change_set.field_names(key));
376 }
377 fields
378 }
379}
380
381#[derive(Debug, Clone, Default)]
382struct EntityMutationLedger {
383 change_sets: ChangeSetStack,
384 comment: Option<String>,
386 deleted_keys: std::collections::BTreeSet<EntityKey>,
389 new_keys: std::collections::BTreeSet<EntityKey>,
391 original_snapshot: Option<OriginalSnapshot>,
393 trace_chains: std::collections::BTreeMap<EntityKey, Vec<teaql_core::TraceNode>>,
395 original_versions: OriginalVersions,
397 composition_errors: Vec<LedgerCompositionError>,
400 is_new: bool,
402}
403
404#[derive(Debug)]
405pub struct EntityRuntimeState {
406 inner: Arc<OnceLock<Arc<Mutex<EntityMutationLedger>>>>,
409 graph: EntityGraphReference,
410 loaded_snapshot: Option<LoadedEntitySnapshot>,
411}
412
413#[derive(Debug, Clone)]
414struct LoadedEntitySnapshot {
415 entity: Arc<str>,
416 row: teaql_core::CompactRow,
417}
418
419#[derive(Debug)]
420enum EntityGraphReference {
421 Strong(Arc<OnceLock<FrozenEntityGraph>>),
422 Weak(Weak<OnceLock<FrozenEntityGraph>>),
423}
424
425impl EntityGraphReference {
426 fn preserve(&self) -> Self {
427 match self {
428 Self::Strong(graph) => Self::Strong(graph.clone()),
429 Self::Weak(graph) => Self::Weak(graph.clone()),
430 }
431 }
432
433 fn promote(&self) -> Self {
434 match self {
435 Self::Strong(graph) => Self::Strong(graph.clone()),
436 Self::Weak(graph) => graph
437 .upgrade()
438 .map(Self::Strong)
439 .unwrap_or_else(|| Self::Strong(Arc::default())),
440 }
441 }
442
443 fn weak(&self) -> Self {
444 match self {
445 Self::Strong(graph) => Self::Weak(Arc::downgrade(graph)),
446 Self::Weak(graph) => Self::Weak(graph.clone()),
447 }
448 }
449
450 fn strong(&self) -> Option<&Arc<OnceLock<FrozenEntityGraph>>> {
451 match self {
452 Self::Strong(graph) => Some(graph),
453 Self::Weak(_) => None,
454 }
455 }
456
457 fn frozen(&self) -> Option<&FrozenEntityGraph> {
458 match self {
459 Self::Strong(graph) => graph.get(),
460 Self::Weak(graph) => {
461 let owner = graph.upgrade()?;
462 let frozen = owner.get()? as *const FrozenEntityGraph;
463 Some(unsafe { &*frozen })
468 }
469 }
470 }
471}
472
473impl Default for EntityRuntimeState {
474 fn default() -> Self {
475 Self {
476 inner: Arc::default(),
477 graph: EntityGraphReference::Strong(Arc::default()),
478 loaded_snapshot: None,
479 }
480 }
481}
482
483impl Clone for EntityRuntimeState {
484 fn clone(&self) -> Self {
485 Self {
486 inner: self.inner.clone(),
487 graph: self.graph.promote(),
488 loaded_snapshot: self.loaded_snapshot.clone(),
489 }
490 }
491}
492
493impl std::panic::UnwindSafe for EntityRuntimeState {}
494impl std::panic::RefUnwindSafe for EntityRuntimeState {}
495
496#[derive(Debug, Clone)]
497enum OriginalSnapshot {
498 Materialized(EntitySnapshot),
499 #[allow(dead_code)] Compact(teaql_core::CompactRow),
501}
502
503impl PartialEq for EntityRuntimeState {
504 fn eq(&self, other: &Self) -> bool {
505 if Arc::ptr_eq(&self.inner, &other.inner) {
506 return true;
507 }
508 match (self.inner.get(), other.inner.get()) {
509 (Some(left), Some(right)) => Arc::ptr_eq(left, right),
510 (None, None) => false,
511 _ => false,
512 }
513 }
514}
515
516impl EntityRuntimeState {
517 #[cfg(test)]
518 fn has_mutation_context(&self) -> bool {
519 self.inner.get().is_some()
520 }
521
522 fn context(&self) -> &Arc<Mutex<EntityMutationLedger>> {
523 self.inner
524 .get_or_init(|| Arc::new(Mutex::new(EntityMutationLedger::default())))
525 }
526
527 fn read_context<R>(&self, default: R, read: impl FnOnce(&EntityMutationLedger) -> R) -> R {
528 let Some(context) = self.inner.get() else {
529 return default;
530 };
531 let context = context.lock().unwrap_or_else(|error| error.into_inner());
532 read(&context)
533 }
534
535 fn write_context<R>(&self, write: impl FnOnce(&mut EntityMutationLedger) -> R) -> R {
536 let mut context = self
537 .context()
538 .lock()
539 .unwrap_or_else(|error| error.into_inner());
540 write(&mut context)
541 }
542
543 pub fn fresh_with_shared_graph(source: &EntityRuntimeState) -> Self {
544 Self {
545 inner: Arc::default(),
546 graph: source.graph.preserve(),
547 loaded_snapshot: None,
548 }
549 }
550
551 pub(crate) fn fresh_with_weak_graph(source: &EntityRuntimeState) -> Self {
554 Self {
555 inner: Arc::default(),
556 graph: source.graph.weak(),
557 loaded_snapshot: None,
558 }
559 }
560
561 pub fn with_shared_graph(&self, source: &EntityRuntimeState) -> Self {
564 Self {
565 inner: self.inner.clone(),
566 graph: source.graph.preserve(),
567 loaded_snapshot: self.loaded_snapshot.clone(),
568 }
569 }
570
571 #[doc(hidden)]
577 pub fn adopt_mutations_from(&self, source: &EntityRuntimeState) {
578 if let Err(error) = self.adopt_mutations_from_checked(source) {
579 self.write_context(|target| target.composition_errors.push(error));
580 }
581 }
582
583 fn adopt_mutations_from_checked(
584 &self,
585 source: &EntityRuntimeState,
586 ) -> Result<(), LedgerCompositionError> {
587 let Some(source_context) = source.inner.get() else {
588 return Ok(());
589 };
590 if self
591 .inner
592 .get()
593 .is_some_and(|target_context| Arc::ptr_eq(target_context, source_context))
594 {
595 return Ok(());
596 }
597 let snapshot = source_context
598 .lock()
599 .unwrap_or_else(|error| error.into_inner())
600 .clone();
601 let loaded_version = source.loaded_snapshot.as_ref().and_then(|loaded| {
602 let id = loaded.row.get("id")?.clone();
603 let version = loaded.row.get("version")?.try_i64()?;
604 Some((EntityKey::new(loaded.entity.as_ref(), id), version))
605 });
606 self.write_context(|target| {
607 let source_versions = snapshot
608 .original_versions
609 .first
610 .iter()
611 .map(|(key, version)| (key, *version))
612 .chain(
613 snapshot
614 .original_versions
615 .overflow
616 .iter()
617 .map(|(key, version)| (key, *version)),
618 )
619 .chain(loaded_version.iter().map(|(key, version)| (key, *version)));
620 for (key, source_version) in source_versions {
621 let target_version = target.original_versions.get(key).or_else(|| {
622 let loaded = self.loaded_snapshot.as_ref()?;
623 (loaded.entity.as_ref() == key.entity.as_ref()
624 && loaded.row.get("id")?.try_u64() == key.id.try_u64())
625 .then(|| loaded.row.get("version")?.try_i64())
626 .flatten()
627 });
628 if let Some(target_version) = target_version
629 && target_version != source_version
630 {
631 return Err(LedgerCompositionError::ConflictingOriginalVersion {
632 entity: key.entity.to_string(),
633 id: key
634 .id
635 .try_u64()
636 .map(|id| id.to_string())
637 .unwrap_or_else(|| format!("{:?}", key.id)),
638 target: target_version,
639 source: source_version,
640 });
641 }
642 }
643 for change_set in snapshot.change_sets.stack {
644 for (key, values) in change_set.changes {
645 for (field, value) in values {
646 target.change_sets.set(key.clone(), field, value);
647 }
648 }
649 }
650 target.deleted_keys.extend(snapshot.deleted_keys);
651 target.new_keys.extend(snapshot.new_keys);
652 target
653 .original_versions
654 .merge_from(&snapshot.original_versions);
655 if let Some((key, version)) = loaded_version {
656 target.original_versions.insert(key, version);
657 }
658 for (key, traces) in snapshot.trace_chains {
659 target.trace_chains.entry(key).or_default().extend(traces);
660 }
661 if target.original_snapshot.is_none() {
662 target.original_snapshot = snapshot.original_snapshot;
663 }
664 if target.comment.is_none() {
665 target.comment = snapshot.comment;
666 }
667 target.is_new |= snapshot.is_new;
668 Ok(())
669 })
670 }
671
672 pub fn freeze_graph(&self, builder: EntityGraphBuilder) -> Result<(), EntityGraphBuilder> {
674 let Some(graph) = self.graph.strong() else {
675 return Err(builder);
676 };
677 graph
678 .set(builder.freeze())
679 .map_err(|graph| EntityGraphBuilder {
680 tables: graph.tables,
681 relation_lists: graph.relation_lists,
682 })
683 }
684
685 pub fn resolve_entity<T>(&self, id: u64) -> Option<&T>
687 where
688 T: Any + Send + Sync,
689 {
690 self.graph
691 .frozen()?
692 .tables
693 .get(&TypeId::of::<T>())?
694 .get(&id)?
695 .downcast_ref::<T>()
696 }
697
698 pub fn resolve_relation_list<T>(
699 &self,
700 owner_entity: &str,
701 owner_id: u64,
702 relation: &str,
703 ) -> Option<&SmartList<T>>
704 where
705 T: Any + Send + Sync,
706 {
707 self.graph
708 .frozen()?
709 .relation_lists
710 .get(&RelationListKey {
711 owner_entity: crate::canonical_id_space_entity(owner_entity),
712 owner_id,
713 relation: relation.to_owned(),
714 })?
715 .downcast_ref::<SmartList<T>>()
716 }
717
718 pub fn relation_list<T>(
720 &self,
721 owner_entity: &str,
722 owner_id: u64,
723 relation: &str,
724 ) -> RelationHandle<'_, SmartList<T>>
725 where
726 T: Any + Send + Sync,
727 {
728 let Some(graph) = self.graph.frozen() else {
729 return RelationHandle::new(LoadedRelation::NotLoaded, None);
730 };
731 let key = RelationListKey {
732 owner_entity: crate::canonical_id_space_entity(owner_entity),
733 owner_id,
734 relation: relation.to_owned(),
735 };
736 let Some(stored) = graph.relation_lists.get(&key) else {
737 return RelationHandle::new(LoadedRelation::NotLoaded, None);
738 };
739 let list = stored.downcast_ref::<SmartList<T>>().unwrap_or_else(|| {
740 panic!(
741 "relation view type mismatch: owner={} id={} relation={}",
742 owner_entity, owner_id, relation
743 )
744 });
745 if list.is_empty() {
746 RelationHandle::new(LoadedRelation::Empty, Some(list))
747 } else {
748 RelationHandle::new(LoadedRelation::Loaded, Some(list))
749 }
750 }
751
752 pub fn resolve_relation_option<T>(
753 &self,
754 owner_entity: &str,
755 owner_id: u64,
756 relation: &str,
757 ) -> Option<&Option<T>>
758 where
759 T: Any + Send + Sync,
760 {
761 self.graph
762 .frozen()?
763 .relation_lists
764 .get(&RelationListKey {
765 owner_entity: crate::canonical_id_space_entity(owner_entity),
766 owner_id,
767 relation: relation.to_owned(),
768 })?
769 .downcast_ref::<Option<T>>()
770 }
771
772 pub fn relation_option<T>(
774 &self,
775 owner_entity: &str,
776 owner_id: u64,
777 relation: &str,
778 ) -> RelationHandle<'_, T>
779 where
780 T: Any + Send + Sync,
781 {
782 let Some(graph) = self.graph.frozen() else {
783 return RelationHandle::new(LoadedRelation::NotLoaded, None);
784 };
785 let key = RelationListKey {
786 owner_entity: crate::canonical_id_space_entity(owner_entity),
787 owner_id,
788 relation: relation.to_owned(),
789 };
790 let Some(stored) = graph.relation_lists.get(&key) else {
791 return RelationHandle::new(LoadedRelation::NotLoaded, None);
792 };
793 let value = stored.downcast_ref::<Option<T>>().unwrap_or_else(|| {
794 panic!(
795 "relation view type mismatch: owner={} id={} relation={}",
796 owner_entity, owner_id, relation
797 )
798 });
799 match value {
800 Some(value) => RelationHandle::new(LoadedRelation::Loaded, Some(value)),
801 None => RelationHandle::new(LoadedRelation::Empty, None),
802 }
803 }
804
805 pub fn has_relation_view(&self, owner_entity: &str, owner_id: u64, relation: &str) -> bool {
806 self.graph.frozen().is_some_and(|graph| {
807 graph.relation_lists.contains_key(&RelationListKey {
808 owner_entity: crate::canonical_id_space_entity(owner_entity),
809 owner_id,
810 relation: relation.to_owned(),
811 })
812 })
813 }
814
815 pub fn push_change_set(&self) {
816 self.write_context(|context| context.change_sets.push());
817 }
818
819 pub fn pop_change_set(&self) -> Option<EntityChangeSet> {
820 self.inner.get()?;
821 self.write_context(|context| context.change_sets.pop())
822 }
823
824 pub fn clear_current_change_set(&self) {
825 if self.inner.get().is_some() {
826 self.write_context(|context| context.change_sets.clear_current());
827 }
828 }
829
830 pub fn clear_committed(&self) {
833 if self.inner.get().is_some() {
834 self.write_context(|context| {
835 context.change_sets = ChangeSetStack::default();
836 context.deleted_keys.clear();
837 context.new_keys.clear();
838 context.original_versions.clear();
839 context.trace_chains.clear();
840 context.original_snapshot = None;
841 context.comment = None;
842 context.is_new = false;
843 });
844 }
845 }
846
847 pub fn set(&self, key: EntityKey, field: impl Into<String>, value: impl Into<Value>) {
848 self.write_context(|context| context.change_sets.set(key, field, value.into()));
849 }
850
851 pub fn get(&self, key: &EntityKey, field: &str) -> Option<Value> {
852 self.read_context(None, |context| context.change_sets.get(key, field))
853 }
854
855 pub fn current_change_set(&self) -> EntityChangeSet {
856 self.read_context(EntityChangeSet::default(), |context| {
857 context.change_sets.current().cloned().unwrap_or_default()
858 })
859 }
860
861 pub fn set_comment(&self, comment: impl Into<String>) {
864 self.write_context(|context| context.comment = Some(comment.into()));
865 }
866
867 pub fn get_comment(&self) -> Option<String> {
869 self.read_context(None, |context| context.comment.clone())
870 }
871
872 pub fn mark_as_new(&self, key: EntityKey) {
874 self.write_context(|context| {
875 context.new_keys.insert(key);
876 });
877 }
878
879 pub fn is_new(&self, key: &EntityKey) -> bool {
881 self.read_context(false, |context| context.new_keys.contains(key))
882 }
883
884 pub fn set_original_snapshot(&self, snapshot: EntitySnapshot) {
886 self.write_context(|context| {
887 context.original_snapshot = Some(OriginalSnapshot::Materialized(snapshot));
888 });
889 }
890
891 pub fn set_original_compact_row(
893 &mut self,
894 entity: impl Into<Arc<str>>,
895 row: teaql_core::CompactRow,
896 ) {
897 self.loaded_snapshot = Some(LoadedEntitySnapshot {
898 entity: entity.into(),
899 row,
900 });
901 }
902
903 pub fn original_snapshot(&self) -> Option<EntitySnapshot> {
905 if let Some(snapshot) = &self.loaded_snapshot {
906 return Some(EntitySnapshot::from(snapshot.row.clone().into_map()));
907 }
908 self.read_context(None, |context| {
909 context
910 .original_snapshot
911 .as_ref()
912 .map(|snapshot| match snapshot {
913 OriginalSnapshot::Materialized(snapshot) => snapshot.clone(),
914 OriginalSnapshot::Compact(row) => EntitySnapshot::from(row.clone().into_map()),
915 })
916 })
917 }
918
919 pub fn mark_as_delete(&self, key: EntityKey) {
924 self.write_context(|context| {
925 context.change_sets.clear_entity(&key);
926 context.deleted_keys.insert(key);
927 });
928 }
929
930 pub fn is_marked_as_delete(&self, key: &EntityKey) -> bool {
932 self.read_context(false, |context| context.deleted_keys.contains(key))
933 }
934
935 pub fn changed_field_names(&self, key: &EntityKey) -> BTreeSet<String> {
938 self.read_context(BTreeSet::new(), |context| {
939 context.change_sets.changed_field_names(key)
940 })
941 }
942 pub fn deleted_keys(&self) -> std::collections::BTreeSet<EntityKey> {
943 self.read_context(BTreeSet::new(), |context| context.deleted_keys.clone())
944 }
945
946 pub fn new_keys(&self) -> std::collections::BTreeSet<EntityKey> {
947 self.read_context(BTreeSet::new(), |context| context.new_keys.clone())
948 }
949
950 pub fn get_original_version(&self, key: &EntityKey) -> Option<i64> {
951 self.read_context(None, |context| context.original_versions.get(key))
952 .or_else(|| {
953 let snapshot = self.loaded_snapshot.as_ref()?;
954 if snapshot.entity.as_ref() != key.entity.as_ref() {
955 return None;
956 }
957 snapshot
958 .row
959 .get("id")?
960 .try_u64()
961 .filter(|id| Some(*id) == key.id.try_u64())?;
962 snapshot.row.get("version")?.try_i64()
963 })
964 }
965
966 pub fn get_trace_chain(&self, key: &EntityKey) -> Vec<teaql_core::TraceNode> {
967 self.read_context(Vec::new(), |context| {
968 context.trace_chains.get(key).cloned().unwrap_or_default()
969 })
970 }
971
972 pub fn set_original_version(&self, key: EntityKey, version: i64) {
973 self.write_context(|context| context.original_versions.insert(key, version));
974 }
975
976 pub(crate) fn first_composition_error(&self) -> Option<LedgerCompositionError> {
977 self.read_context(None, |context| context.composition_errors.first().cloned())
978 }
979}
980
981#[derive(Clone, Debug, Eq, PartialEq)]
983pub enum LedgerCompositionError {
984 MissingTargetState,
985 MissingSourceState,
986 ConflictingOriginalVersion {
987 entity: String,
988 id: String,
989 target: i64,
990 source: i64,
991 },
992}
993
994impl std::fmt::Display for LedgerCompositionError {
995 fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
996 match self {
997 Self::MissingTargetState => formatter.write_str("target entity has no mutation ledger"),
998 Self::MissingSourceState => formatter.write_str("source entity has no mutation ledger"),
999 Self::ConflictingOriginalVersion {
1000 entity,
1001 id,
1002 target,
1003 source,
1004 } => write!(
1005 formatter,
1006 "cannot compose {entity}#{id}: receiving ledger expects version {target}, source expects version {source}"
1007 ),
1008 }
1009 }
1010}
1011
1012impl std::error::Error for LedgerCompositionError {}
1013
1014pub trait LedgerEntity: teaql_core::Entity {
1015 fn entity_runtime_state(&self) -> Option<EntityRuntimeState>;
1016
1017 fn include_pending_mutations_from<E: LedgerEntity>(
1023 &self,
1024 source: &E,
1025 ) -> Result<(), LedgerCompositionError> {
1026 let target = self
1027 .entity_runtime_state()
1028 .ok_or(LedgerCompositionError::MissingTargetState)?;
1029 let source = source
1030 .entity_runtime_state()
1031 .ok_or(LedgerCompositionError::MissingSourceState)?;
1032 target.adopt_mutations_from_checked(&source)
1033 }
1034}
1035
1036#[cfg(test)]
1037mod composition_api_tests {
1038 use super::*;
1039 use teaql_core::{CompactRow, EntityDescriptor, EntityError, MutationValues, TeaqlEntity};
1040
1041 #[test]
1042 fn numeric_id_keys_match_across_sqlite_signed_and_entity_unsigned_values() {
1043 let signed = EntityKey::new_static("Child", Value::I64(1));
1044 let unsigned = EntityKey::new_static("Child", Value::U64(1));
1045 assert_eq!(signed, unsigned);
1046 assert_eq!(signed.cmp(&unsigned), std::cmp::Ordering::Equal);
1047 assert_ne!(signed, EntityKey::new_static("OtherChild", Value::U64(1)));
1048 assert_ne!(
1049 EntityKey::new_static("Child", Value::I64(-1)),
1050 EntityKey::new_static("Child", Value::U64(u64::MAX))
1051 );
1052 }
1053
1054 struct TestEntity(Option<EntityRuntimeState>);
1055
1056 impl TeaqlEntity for TestEntity {
1057 const ENTITY_NAME: &'static str = "TestEntity";
1058
1059 fn entity_descriptor() -> EntityDescriptor {
1060 EntityDescriptor::new(Self::ENTITY_NAME)
1061 }
1062 }
1063
1064 impl teaql_core::Entity for TestEntity {
1065 fn from_compact_row(_row: CompactRow) -> Result<Self, EntityError> {
1066 Err(EntityError::new(Self::ENTITY_NAME, "not used by this test"))
1067 }
1068
1069 fn into_values(self) -> MutationValues {
1070 MutationValues::new()
1071 }
1072 }
1073
1074 impl LedgerEntity for TestEntity {
1075 fn entity_runtime_state(&self) -> Option<EntityRuntimeState> {
1076 self.0.clone()
1077 }
1078 }
1079
1080 #[test]
1081 fn public_composition_reports_missing_ledgers_without_silent_success() {
1082 let present = TestEntity(Some(EntityRuntimeState::default()));
1083 let missing = TestEntity(None);
1084 assert_eq!(
1085 missing.include_pending_mutations_from(&present),
1086 Err(LedgerCompositionError::MissingTargetState)
1087 );
1088 assert_eq!(
1089 present.include_pending_mutations_from(&missing),
1090 Err(LedgerCompositionError::MissingSourceState)
1091 );
1092 }
1093
1094 #[test]
1095 fn public_composition_copies_intent_without_sharing_source_ledger() {
1096 let target_state = EntityRuntimeState::default();
1097 let source_state = EntityRuntimeState::default();
1098 let key = EntityKey::new_static("Child", 42_u64);
1099 source_state.set(key.clone(), "name", Value::Text("first".to_owned()));
1100 let target = TestEntity(Some(target_state.clone()));
1101 let source = TestEntity(Some(source_state.clone()));
1102
1103 target.include_pending_mutations_from(&source).unwrap();
1104 assert_eq!(
1105 target_state.get(&key, "name"),
1106 Some(Value::Text("first".to_owned()))
1107 );
1108 source_state.set(key.clone(), "name", Value::Text("second".to_owned()));
1109 assert_eq!(
1110 target_state.get(&key, "name"),
1111 Some(Value::Text("first".to_owned()))
1112 );
1113 target.include_pending_mutations_from(&source).unwrap();
1114 assert_eq!(
1115 target_state.get(&key, "name"),
1116 Some(Value::Text("second".to_owned()))
1117 );
1118 }
1119
1120 #[test]
1121 fn public_composition_retains_sqlite_signed_snapshot_version_for_unsigned_entity_id() {
1122 let target_state = EntityRuntimeState::default();
1123 let mut source_state = EntityRuntimeState::default();
1124 source_state.set_original_compact_row(
1125 "Child",
1126 CompactRow::new(
1127 Arc::from(["id".to_owned(), "version".to_owned()]),
1128 vec![Value::I64(1), Value::I64(7)],
1129 ),
1130 );
1131 let entity_key = EntityKey::new_static("Child", Value::U64(1));
1132 source_state.set(entity_key.clone(), "quantity", Value::I64(2));
1133 TestEntity(Some(target_state.clone()))
1134 .include_pending_mutations_from(&TestEntity(Some(source_state)))
1135 .unwrap();
1136 assert_eq!(target_state.get_original_version(&entity_key), Some(7));
1137 }
1138
1139 #[test]
1140 fn public_composition_rejects_conflicting_versions_without_copying_intent() {
1141 let mut target_state = EntityRuntimeState::default();
1142 let mut source_state = EntityRuntimeState::default();
1143 let unsigned_key = EntityKey::new_static("Child", Value::U64(1));
1144 let schema: Arc<[String]> = Arc::from(["id".to_owned(), "version".to_owned()]);
1145 target_state.set_original_compact_row(
1146 "Child",
1147 CompactRow::new(schema.clone(), vec![Value::I64(1), Value::I64(2)]),
1148 );
1149 target_state.set(unsigned_key.clone(), "quantity", Value::I64(4));
1150 source_state.set_original_compact_row(
1151 "Child",
1152 CompactRow::new(schema, vec![Value::I64(1), Value::I64(1)]),
1153 );
1154 source_state.set(unsigned_key.clone(), "quantity", Value::I64(3));
1155
1156 let result = TestEntity(Some(target_state.clone()))
1157 .include_pending_mutations_from(&TestEntity(Some(source_state.clone())));
1158 assert_eq!(
1159 result,
1160 Err(LedgerCompositionError::ConflictingOriginalVersion {
1161 entity: "Child".to_owned(),
1162 id: "1".to_owned(),
1163 target: 2,
1164 source: 1,
1165 })
1166 );
1167 assert_eq!(target_state.get_original_version(&unsigned_key), Some(2));
1168 assert_eq!(
1169 target_state.get(&unsigned_key, "quantity"),
1170 Some(Value::I64(4))
1171 );
1172 assert_eq!(source_state.get_original_version(&unsigned_key), Some(1));
1173 assert_eq!(
1174 source_state.get(&unsigned_key, "quantity"),
1175 Some(Value::I64(3))
1176 );
1177 }
1178
1179 #[test]
1180 fn public_composition_accepts_equal_versions_and_distinct_entity_keys() {
1181 let target_state = EntityRuntimeState::default();
1182 let source_state = EntityRuntimeState::default();
1183 let same_key = EntityKey::new_static("Child", 1_u64);
1184 let other_key = EntityKey::new_static("OtherChild", 1_u64);
1185 target_state.set_original_version(same_key.clone(), 2);
1186 source_state.set_original_version(same_key.clone(), 2);
1187 source_state.set_original_version(other_key.clone(), 9);
1188 source_state.set(same_key.clone(), "quantity", Value::I64(3));
1189 source_state.set(other_key.clone(), "name", Value::Text("other".to_owned()));
1190
1191 TestEntity(Some(target_state.clone()))
1192 .include_pending_mutations_from(&TestEntity(Some(source_state)))
1193 .unwrap();
1194 assert_eq!(target_state.get_original_version(&same_key), Some(2));
1195 assert_eq!(target_state.get_original_version(&other_key), Some(9));
1196 assert_eq!(target_state.get(&same_key, "quantity"), Some(Value::I64(3)));
1197 assert_eq!(
1198 target_state.get(&other_key, "name"),
1199 Some(Value::Text("other".to_owned()))
1200 );
1201 }
1202
1203 #[test]
1204 fn generated_void_attachment_retains_version_conflict_for_save_preflight() {
1205 let target_state = EntityRuntimeState::default();
1206 let source_state = EntityRuntimeState::default();
1207 let key = EntityKey::new_static("Child", 1_u64);
1208 target_state.set_original_version(key.clone(), 2);
1209 source_state.set_original_version(key.clone(), 1);
1210 source_state.set(key.clone(), "quantity", Value::I64(3));
1211
1212 target_state.adopt_mutations_from(&source_state);
1213 assert_eq!(
1214 target_state.first_composition_error(),
1215 Some(LedgerCompositionError::ConflictingOriginalVersion {
1216 entity: "Child".to_owned(),
1217 id: "1".to_owned(),
1218 target: 2,
1219 source: 1,
1220 })
1221 );
1222 assert_eq!(target_state.get_original_version(&key), Some(2));
1223 assert_eq!(target_state.get(&key, "quantity"), None);
1224 assert_eq!(source_state.get(&key, "quantity"), Some(Value::I64(3)));
1225 }
1226}
1227
1228#[cfg(test)]
1229mod lazy_root_tests {
1230 use super::*;
1231
1232 #[derive(Clone)]
1233 struct GraphChild {
1234 root: EntityRuntimeState,
1235 }
1236
1237 #[test]
1238 fn loaded_snapshot_does_not_allocate_ledger_until_mutation() {
1239 let mut root = EntityRuntimeState::default();
1240 root.set_original_compact_row(
1241 "Example",
1242 teaql_core::CompactRow::new(
1243 Arc::from(["id".to_owned(), "version".to_owned()]),
1244 vec![Value::U64(7), Value::I64(3)],
1245 ),
1246 );
1247 let key = EntityKey::new_static("Example", 7_u64);
1248
1249 assert!(!root.has_mutation_context());
1250 assert_eq!(root.get(&key, "name"), None);
1251 assert_eq!(root.get_original_version(&key), Some(3));
1252 assert!(!root.has_mutation_context());
1253
1254 root.set(key, "name", Value::Text("updated".to_owned()));
1255 assert!(root.has_mutation_context());
1256 }
1257
1258 #[test]
1259 fn graph_composition_retains_version_for_same_id_across_entity_types() {
1260 let mut parent = EntityRuntimeState::default();
1261 parent.set_original_compact_row(
1262 "Order",
1263 teaql_core::CompactRow::new(
1264 Arc::from(["id".to_owned(), "version".to_owned()]),
1265 vec![Value::U64(1), Value::I64(1)],
1266 ),
1267 );
1268 let mut execution = EntityRuntimeState::default();
1269 execution.set_original_compact_row(
1270 "InferenceExecution",
1271 teaql_core::CompactRow::new(
1272 Arc::from(["id".to_owned(), "version".to_owned()]),
1273 vec![Value::U64(1), Value::I64(2)],
1274 ),
1275 );
1276 let order_key = EntityKey::new_static("Order", 1_u64);
1277 let execution_key = EntityKey::new_static("InferenceExecution", 1_u64);
1278 execution.set(
1279 execution_key.clone(),
1280 "execution_status",
1281 Value::Text("COMPLETED".to_owned()),
1282 );
1283
1284 assert_eq!(parent.get_original_version(&execution_key), None);
1285 parent.adopt_mutations_from(&execution);
1286
1287 assert_eq!(parent.get_original_version(&order_key), Some(1));
1288 assert_eq!(parent.get_original_version(&execution_key), Some(2));
1289 }
1290
1291 #[test]
1292 fn clone_before_first_mutation_materializes_one_shared_ledger() {
1293 let root = EntityRuntimeState::default();
1294 let child = root.clone();
1295 let child_key = EntityKey::new_static("Child", 2_u64);
1296
1297 child.set(child_key.clone(), "name", Value::Text("updated".to_owned()));
1298
1299 assert_eq!(
1300 root.get(&child_key, "name"),
1301 Some(Value::Text("updated".to_owned()))
1302 );
1303 assert!(root.has_mutation_context());
1304 }
1305
1306 #[test]
1307 fn explicit_graph_composition_adopts_existing_child_mutations() {
1308 let parent = EntityRuntimeState::default();
1309 let child = EntityRuntimeState::default();
1310 let child_key = EntityKey::new_static("Child", 17_u64);
1311
1312 child.mark_as_new(child_key.clone());
1313 child.set(child_key.clone(), "display_name", "before attach");
1314 child.set_original_version(child_key.clone(), 3);
1315
1316 parent.adopt_mutations_from(&child);
1317
1318 assert!(parent.is_new(&child_key));
1319 assert_eq!(
1320 parent.get(&child_key, "display_name"),
1321 Some(Value::Text("before attach".to_owned()))
1322 );
1323 assert_eq!(parent.get_original_version(&child_key), Some(3));
1324 }
1325
1326 #[test]
1327 fn graph_owned_entities_do_not_keep_the_graph_alive() {
1328 let root = EntityRuntimeState::default();
1329 let graph_owner = match &root.graph {
1330 EntityGraphReference::Strong(graph) => Arc::downgrade(graph),
1331 EntityGraphReference::Weak(_) => unreachable!(),
1332 };
1333 let mut builder = EntityGraphBuilder::default();
1334 builder.install_relation_list(
1335 "Owner",
1336 1,
1337 "children",
1338 SmartList::from(vec![GraphChild {
1339 root: EntityRuntimeState::fresh_with_weak_graph(&root),
1340 }]),
1341 );
1342 root.freeze_graph(builder).unwrap();
1343
1344 drop(root);
1345 assert!(graph_owner.upgrade().is_none());
1346 }
1347
1348 #[test]
1349 fn cloning_a_graph_owned_entity_promotes_its_graph_anchor() {
1350 let root = EntityRuntimeState::default();
1351 let graph_owner = match &root.graph {
1352 EntityGraphReference::Strong(graph) => Arc::downgrade(graph),
1353 EntityGraphReference::Weak(_) => unreachable!(),
1354 };
1355 let mut builder = EntityGraphBuilder::default();
1356 builder.install_relation_list(
1357 "Owner",
1358 1,
1359 "children",
1360 SmartList::from(vec![GraphChild {
1361 root: EntityRuntimeState::fresh_with_weak_graph(&root),
1362 }]),
1363 );
1364 root.freeze_graph(builder).unwrap();
1365 let detached = root
1366 .resolve_relation_list::<GraphChild>("Owner", 1, "children")
1367 .unwrap()[0]
1368 .clone();
1369
1370 drop(root);
1371 assert!(graph_owner.upgrade().is_some());
1372 assert!(detached.root.graph.frozen().is_some());
1373 drop(detached);
1374 assert!(graph_owner.upgrade().is_none());
1375 }
1376
1377 #[test]
1378 fn relation_handles_distinguish_loaded_empty_and_not_loaded() {
1379 let root = EntityRuntimeState::default();
1380 let mut builder = EntityGraphBuilder::default();
1381 builder.install_relation_list("Owner", 1, "loaded", SmartList::from(vec![7_u64]));
1382 builder.install_relation_list::<u64>("Owner", 1, "empty", SmartList::empty());
1383 builder.install_relation_option("Owner", 1, "present", Some(9_u64));
1384 builder.install_relation_option::<u64>("Owner", 1, "null", None);
1385 root.freeze_graph(builder).unwrap();
1386
1387 let loaded = root.relation_list::<u64>("Owner", 1, "loaded");
1388 assert_eq!(loaded.state(), LoadedRelation::Loaded);
1389 assert_eq!(loaded.value().map(|list| list.as_slice()), Some(&[7][..]));
1390
1391 let empty = root.relation_list::<u64>("Owner", 1, "empty");
1392 assert_eq!(empty.state(), LoadedRelation::Empty);
1393 assert!(empty.value().is_some_and(SmartList::is_empty));
1394
1395 let missing = root.relation_list::<u64>("Owner", 1, "missing");
1396 assert_eq!(missing.state(), LoadedRelation::NotLoaded);
1397 assert!(missing.value().is_none());
1398
1399 let present = root.relation_option::<u64>("Owner", 1, "present");
1400 assert_eq!(present.state(), LoadedRelation::Loaded);
1401 assert_eq!(present.value(), Some(&9));
1402
1403 let null = root.relation_option::<u64>("Owner", 1, "null");
1404 assert_eq!(null.state(), LoadedRelation::Empty);
1405 assert!(null.value().is_none());
1406
1407 let absent = root.relation_option::<u64>("Owner", 1, "absent");
1408 assert_eq!(absent.state(), LoadedRelation::NotLoaded);
1409 assert!(absent.value().is_none());
1410 }
1411}