Skip to main content

teaql_runtime/
graph.rs

1use std::collections::{BTreeMap, BTreeSet};
2use std::ops::{Deref, DerefMut};
3use std::sync::Arc;
4
5use teaql_core::{EntitySnapshot, MutationValues, TraceNode, Value};
6
7/// Mutable field state for one entity while checker/fix and graph planning run.
8/// It is deliberately distinct from query rows, mutation commands, and loaded
9/// snapshots.
10#[derive(Debug, Clone, Default, PartialEq)]
11pub struct EntityValues(BTreeMap<String, Value>);
12
13impl EntityValues {
14    pub fn new() -> Self {
15        Self::default()
16    }
17}
18
19impl Deref for EntityValues {
20    type Target = BTreeMap<String, Value>;
21
22    fn deref(&self) -> &Self::Target {
23        &self.0
24    }
25}
26
27impl DerefMut for EntityValues {
28    fn deref_mut(&mut self) -> &mut Self::Target {
29        &mut self.0
30    }
31}
32
33impl From<BTreeMap<String, Value>> for EntityValues {
34    fn from(values: BTreeMap<String, Value>) -> Self {
35        Self(values)
36    }
37}
38
39impl From<EntityValues> for BTreeMap<String, Value> {
40    fn from(values: EntityValues) -> Self {
41        values.0
42    }
43}
44
45impl From<EntityValues> for MutationValues {
46    fn from(values: EntityValues) -> Self {
47        BTreeMap::from(values).into()
48    }
49}
50
51impl From<MutationValues> for EntityValues {
52    fn from(values: MutationValues) -> Self {
53        let values: BTreeMap<String, Value> = values.into();
54        values.into()
55    }
56}
57
58impl From<teaql_core::CompactRow> for EntityValues {
59    fn from(row: teaql_core::CompactRow) -> Self {
60        row.into_map().into()
61    }
62}
63
64impl IntoIterator for EntityValues {
65    type Item = (String, Value);
66    type IntoIter = std::collections::btree_map::IntoIter<String, Value>;
67
68    fn into_iter(self) -> Self::IntoIter {
69        self.0.into_iter()
70    }
71}
72
73impl<'a> IntoIterator for &'a EntityValues {
74    type Item = (&'a String, &'a Value);
75    type IntoIter = std::collections::btree_map::Iter<'a, String, Value>;
76
77    fn into_iter(self) -> Self::IntoIter {
78        self.0.iter()
79    }
80}
81
82#[derive(Debug, Clone, Copy, PartialEq, Eq)]
83pub enum GraphOperation {
84    Upsert,
85    Create,
86    Reference,
87    Remove,
88}
89
90#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord)]
91pub enum GraphMutationKind {
92    Create,
93    Update,
94    Delete,
95    Reference,
96}
97
98impl GraphMutationKind {
99    pub fn for_update(is_update: bool) -> Self {
100        match is_update {
101            true => Self::Update,
102            false => Self::Create,
103        }
104    }
105}
106
107/// A persistent linked-list token for hierarchical trace context.
108///
109/// Each token holds the trace info for one graph node and an `Arc` pointer
110/// to its parent's token. The full trace chain is only materialized when
111/// explicitly requested via [`recover_trace_chain()`], giving us zero-cost
112/// propagation during the flatten phase.
113#[derive(Debug, Clone, PartialEq)]
114pub struct TraceScopeToken {
115    /// Shared pointer to the parent scope (zero-copy link).
116    pub parent: Option<Arc<TraceScopeToken>>,
117    /// The trace metadata for this scope level.
118    pub track: TraceNode,
119    /// The item_index of the PlanItem that created this scope (for debugging).
120    pub node_index: u64,
121}
122
123impl TraceScopeToken {
124    /// Lazily recover the full trace chain by walking the parent pointers.
125    /// Only called when an event consumer actually needs the chain.
126    pub fn recover_trace_chain(&self) -> Vec<TraceNode> {
127        let mut chain = Vec::new();
128        let mut current: Option<&TraceScopeToken> = Some(self);
129        while let Some(token) = current {
130            if !token.track.comment.is_empty() {
131                chain.push(token.track.clone());
132            }
133            current = token.parent.as_deref();
134        }
135        chain.reverse();
136        chain
137    }
138}
139
140#[derive(Debug, Clone, PartialEq)]
141pub struct GraphMutationPlanItem {
142    pub entity: String,
143    pub kind: GraphMutationKind,
144    pub values: MutationValues,
145    pub update_fields: Vec<String>,
146    /// Monotonically increasing index assigned at push time (for debugging).
147    pub item_index: u64,
148    /// Lazy trace context — only materialized into a Vec<TraceNode> on demand.
149    pub scope_token: Option<Arc<TraceScopeToken>>,
150    pub old_values: Option<EntitySnapshot>,
151}
152
153#[derive(Debug, Clone, PartialEq)]
154pub struct GraphMutationBatch {
155    pub entity: String,
156    pub kind: GraphMutationKind,
157    pub update_fields: Vec<String>,
158    pub items: Vec<GraphMutationPlanItem>,
159}
160
161#[derive(Debug, Clone, PartialEq, Default)]
162pub struct GraphMutationPlan {
163    pub planned_root: Option<GraphNode>,
164    pub items: Vec<GraphMutationPlanItem>,
165    pub batches: Vec<GraphMutationBatch>,
166    /// Auto-incrementing counter for item_index assignment.
167    pub next_item_index: u64,
168    /// Keep track of visited nodes to avoid infinite loops and redundant updates
169    pub visited_nodes: std::collections::HashSet<(String, String)>,
170}
171
172impl GraphMutationPlan {
173    pub fn push(
174        &mut self,
175        entity: impl Into<String>,
176        kind: GraphMutationKind,
177        values: MutationValues,
178        update_fields: Vec<String>,
179        scope_token: Option<Arc<TraceScopeToken>>,
180        old_values: Option<EntitySnapshot>,
181    ) {
182        let index = self.next_item_index;
183        self.next_item_index += 1;
184        self.items.push(GraphMutationPlanItem {
185            entity: entity.into(),
186            kind,
187            values,
188            update_fields,
189            item_index: index,
190            scope_token,
191            old_values,
192        });
193    }
194
195    pub fn rebuild_batches(&mut self) {
196        let mut grouped: BTreeMap<
197            (String, GraphMutationKind, Vec<String>),
198            Vec<GraphMutationPlanItem>,
199        > = BTreeMap::new();
200        for item in &self.items {
201            let update_fields = match item.kind {
202                GraphMutationKind::Update => item.update_fields.clone(),
203                _ => Vec::new(),
204            };
205            grouped
206                .entry((item.entity.clone(), item.kind, update_fields))
207                .or_default()
208                .push(item.clone());
209        }
210        self.batches = grouped
211            .into_iter()
212            .map(
213                |((entity, kind, update_fields), items)| GraphMutationBatch {
214                    entity,
215                    kind,
216                    update_fields,
217                    items,
218                },
219            )
220            .collect();
221    }
222
223    pub fn grouped_counts(&self) -> BTreeMap<(String, GraphMutationKind), usize> {
224        let mut counts = BTreeMap::new();
225        for batch in &self.batches {
226            *counts
227                .entry((batch.entity.clone(), batch.kind))
228                .or_insert(0) += batch.items.len();
229        }
230        counts
231    }
232
233    pub fn batch_count(&self) -> usize {
234        self.batches.len()
235    }
236
237    pub fn len(&self) -> usize {
238        self.items.len()
239    }
240
241    pub fn is_empty(&self) -> bool {
242        self.items.is_empty()
243    }
244}
245
246pub fn sorted_update_fields(
247    values: &EntityValues,
248    excluded: impl IntoIterator<Item = String>,
249) -> Vec<String> {
250    let excluded = excluded.into_iter().collect::<BTreeSet<_>>();
251    values
252        .keys()
253        .filter(|field| !excluded.contains(*field))
254        .cloned()
255        .collect()
256}
257
258#[derive(Debug, Clone, PartialEq)]
259pub struct GraphNode {
260    pub entity: String,
261    pub values: EntityValues,
262    pub relations: BTreeMap<String, Vec<GraphNode>>,
263    pub operation: GraphOperation,
264    /// Annotation comment: carries business intent metadata through graph save.
265    /// Not persisted to the database — used for observability (SQL logs, audit trails).
266    pub comment: Option<String>,
267    /// Fields modified via `update_*()` methods (dirty tracking).
268    /// `None` = all fields (new entity or no tracking available).
269    /// `Some(set)` = only these fields were modified — UPDATE should only include them.
270    /// This is the Rust equivalent of Java's `entity.getUpdatedProperties()`.
271    pub dirty_fields: Option<BTreeSet<String>>,
272    /// L1 Cache snapshot of the entity values exactly as they were loaded from the database.
273    /// Used by the Event Engine to eliminate redundant old_value queries during auditing.
274    pub original_values: Option<EntitySnapshot>,
275}
276
277impl GraphNode {
278    pub fn new(entity: impl Into<String>) -> Self {
279        Self {
280            entity: entity.into(),
281            values: EntityValues::new(),
282            relations: BTreeMap::new(),
283            operation: GraphOperation::Upsert,
284            comment: None,
285            dirty_fields: None,
286            original_values: None,
287        }
288    }
289
290    pub fn operation(mut self, operation: GraphOperation) -> Self {
291        self.operation = operation;
292        self
293    }
294
295    pub fn reference(mut self) -> Self {
296        self.operation = GraphOperation::Reference;
297        self
298    }
299
300    pub fn remove(mut self) -> Self {
301        self.operation = GraphOperation::Remove;
302        self
303    }
304
305    pub fn value(mut self, field: impl Into<String>, value: impl Into<Value>) -> Self {
306        self.values.insert(field.into(), value.into());
307        self
308    }
309
310    pub fn relation(mut self, name: impl Into<String>, node: GraphNode) -> Self {
311        self.relations.entry(name.into()).or_default().push(node);
312        self
313    }
314
315    pub fn relations(
316        mut self,
317        name: impl Into<String>,
318        nodes: impl IntoIterator<Item = GraphNode>,
319    ) -> Self {
320        self.relations.entry(name.into()).or_default().extend(nodes);
321        self
322    }
323
324    pub fn id(&self) -> Option<&Value> {
325        self.values.get("id")
326    }
327
328    /// Set an annotation comment on this graph node.
329    /// The comment propagates through the graph save process for observability.
330    pub fn comment(mut self, comment: impl Into<String>) -> Self {
331        self.comment = Some(comment.into());
332        self
333    }
334
335    /// Set an annotation comment by mutable reference.
336    pub fn set_comment(&mut self, comment: impl Into<String>) {
337        self.comment = Some(comment.into());
338    }
339}
340
341// ---------------------------------------------------------------------------
342// Hierarchical Comment Propagation (Scoped Cons List)
343// ---------------------------------------------------------------------------
344
345/// A stack-allocated scope node forming a parent-pointer cons list.
346///
347/// Each node lives on the call stack of the recursive graph save function.
348/// Child nodes hold a `&'a` reference to their parent's stack frame,
349/// giving us thread-safe, lock-free, zero-overhead hierarchical comment tracking.
350#[derive(Debug)]
351pub struct ScopedCommentNode<'a> {
352    /// Reference to the parent scope (lives on the caller's stack frame)
353    pub parent: Option<&'a ScopedCommentNode<'a>>,
354    pub track: teaql_core::TraceNode,
355}
356
357impl<'a> ScopedCommentNode<'a> {
358    pub fn to_trace_chain(&self) -> Vec<teaql_core::TraceNode> {
359        let mut chain = Vec::new();
360        let mut current: Option<&ScopedCommentNode<'_>> = Some(self);
361
362        while let Some(node) = current {
363            if !node.track.comment.is_empty() {
364                chain.push(node.track.clone());
365            }
366            current = node.parent;
367        }
368
369        chain.reverse();
370        chain
371    }
372}
373
374#[cfg(test)]
375mod tests {
376    use super::*;
377
378    #[test]
379    fn test_hierarchical_trace_chain_recovery() {
380        let root_trace = TraceNode {
381            entity_type: "User".to_string(),
382            entity_id: Some(1),
383            comment: "Create User".to_string(),
384        };
385
386        let child_trace = TraceNode {
387            entity_type: "Profile".to_string(),
388            entity_id: None,
389            comment: "Create Profile".to_string(),
390        };
391
392        let empty_comment_trace = TraceNode {
393            entity_type: "AuditLog".to_string(),
394            entity_id: None,
395            comment: "".to_string(),
396        };
397
398        // Test ScopedCommentNode
399        let root_scope = ScopedCommentNode {
400            parent: None,
401            track: root_trace.clone(),
402        };
403        let child_scope = ScopedCommentNode {
404            parent: Some(&root_scope),
405            track: child_trace.clone(),
406        };
407        let empty_scope = ScopedCommentNode {
408            parent: Some(&child_scope),
409            track: empty_comment_trace.clone(),
410        };
411
412        let chain = empty_scope.to_trace_chain();
413        assert_eq!(chain.len(), 2);
414        assert_eq!(chain[0], root_trace);
415        assert_eq!(chain[1], child_trace);
416
417        // Test TraceScopeToken
418        let root_token = Arc::new(TraceScopeToken {
419            parent: None,
420            track: root_trace.clone(),
421            node_index: 0,
422        });
423        let child_token = Arc::new(TraceScopeToken {
424            parent: Some(root_token),
425            track: child_trace.clone(),
426            node_index: 1,
427        });
428        let empty_token = Arc::new(TraceScopeToken {
429            parent: Some(child_token),
430            track: empty_comment_trace,
431            node_index: 2,
432        });
433
434        let chain = empty_token.recover_trace_chain();
435        assert_eq!(chain.len(), 2);
436        assert_eq!(chain[0], root_trace);
437        assert_eq!(chain[1], child_trace);
438    }
439
440    #[test]
441    fn test_graph_mutation_plan_batching_keys_and_counts() {
442        let mut plan = GraphMutationPlan::default();
443
444        // Push 2 creates for User
445        plan.push(
446            "User",
447            GraphMutationKind::Create,
448            MutationValues::new(),
449            vec![],
450            None,
451            None,
452        );
453        plan.push(
454            "User",
455            GraphMutationKind::Create,
456            MutationValues::new(),
457            vec![],
458            None,
459            None,
460        );
461
462        // Push 2 updates for User with same fields
463        plan.push(
464            "User",
465            GraphMutationKind::Update,
466            MutationValues::new(),
467            vec!["name".to_string()],
468            None,
469            None,
470        );
471        plan.push(
472            "User",
473            GraphMutationKind::Update,
474            MutationValues::new(),
475            vec!["name".to_string()],
476            None,
477            None,
478        );
479
480        // Push 1 update for User with different fields (should be separate batch)
481        plan.push(
482            "User",
483            GraphMutationKind::Update,
484            MutationValues::new(),
485            vec!["email".to_string()],
486            None,
487            None,
488        );
489
490        // Push 1 create for Profile
491        plan.push(
492            "Profile",
493            GraphMutationKind::Create,
494            MutationValues::new(),
495            vec![],
496            None,
497            None,
498        );
499
500        assert_eq!(plan.len(), 6);
501
502        // Rebuild batches
503        plan.rebuild_batches();
504
505        // We expect 4 batches:
506        // 1. User Create (2 items)
507        // 2. User Update ["email"] (1 item)
508        // 3. User Update ["name"] (2 items)
509        // 4. Profile Create (1 item)
510        assert_eq!(plan.batch_count(), 4);
511
512        let counts = plan.grouped_counts();
513        assert_eq!(counts.len(), 3);
514        assert_eq!(
515            counts.get(&("User".to_string(), GraphMutationKind::Create)),
516            Some(&2)
517        );
518        assert_eq!(
519            counts.get(&("User".to_string(), GraphMutationKind::Update)),
520            Some(&3)
521        );
522        assert_eq!(
523            counts.get(&("Profile".to_string(), GraphMutationKind::Create)),
524            Some(&1)
525        );
526    }
527}