teaql_runtime/
graph.rs

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

teaql_runtime/
graph.rs
