arete_server/
materialized_view.rs

1//! Materialized view evaluation for view pipelines.
2//!
3//! This module handles the runtime evaluation of ViewDef pipelines,
4//! maintaining materialized results that update as source data changes.
5
6use crate::cache::EntityCache;
7use serde::Serialize;
8use serde_json::Value;
9use std::collections::{HashMap, HashSet};
10use std::sync::Arc;
11use tokio::sync::RwLock;
12
13/// Result of evaluating whether an update affects a materialized view
14#[derive(Debug, Clone, PartialEq)]
15pub enum ViewEffect {
16    /// Update does not affect the view result
17    NoEffect,
18    /// Entity should be added to the view result
19    Add { key: String },
20    /// Entity should be removed from the view result
21    Remove { key: String },
22    /// Entity in view was updated
23    Update { key: String },
24    /// Entity replaces another in the view (for single-result views)
25    Replace { old_key: String, new_key: String },
26}
27
28/// Sort order for view evaluation
29#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)]
30pub enum SortOrder {
31    Asc,
32    Desc,
33}
34
35/// Comparison operators
36#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)]
37pub enum CompareOp {
38    Eq,
39    Ne,
40    Gt,
41    Gte,
42    Lt,
43    Lte,
44}
45
46/// A materialized view that tracks a subset of entities based on a pipeline
47#[derive(Debug)]
48pub struct MaterializedView {
49    /// View identifier
50    pub id: String,
51    /// Source view/entity this derives from
52    pub source_id: String,
53    /// Current set of entity keys in this view's result
54    current_keys: Arc<RwLock<HashSet<String>>>,
55    /// Pipeline configuration (simplified for now)
56    pipeline: ViewPipeline,
57}
58
59#[derive(Debug, Clone, Default, Serialize)]
60pub struct ViewPipeline {
61    /// Filter predicate (field path, op, value)
62    pub filter: Option<FilterConfig>,
63    /// Sort configuration
64    pub sort: Option<SortConfig>,
65    /// Limit (take N) - if Some(1), treated as single-result view for Replace effects
66    pub limit: Option<usize>,
67}
68
69#[derive(Debug, Clone, Serialize)]
70pub struct FilterConfig {
71    pub field_path: Vec<String>,
72    pub op: CompareOp,
73    pub value: Value,
74}
75
76#[derive(Debug, Clone, Serialize)]
77pub struct SortConfig {
78    pub field_path: Vec<String>,
79    pub order: SortOrder,
80}
81
82impl MaterializedView {
83    /// Create a new materialized view
84    pub fn new(id: String, source_id: String, pipeline: ViewPipeline) -> Self {
85        Self {
86            id,
87            source_id,
88            current_keys: Arc::new(RwLock::new(HashSet::new())),
89            pipeline,
90        }
91    }
92
93    /// Get current keys in the view
94    pub async fn get_keys(&self) -> HashSet<String> {
95        self.current_keys.read().await.clone()
96    }
97
98    /// Evaluate initial state from cache
99    pub async fn evaluate_initial(&self, cache: &EntityCache) -> Vec<(String, Value)> {
100        let entities = cache.get_all(&self.source_id).await;
101        self.evaluate_pipeline(entities).await
102    }
103
104    /// Evaluate pipeline on a set of entities
105    async fn evaluate_pipeline(&self, mut entities: Vec<(String, Value)>) -> Vec<(String, Value)> {
106        // Apply filter
107        if let Some(ref filter) = self.pipeline.filter {
108            entities.retain(|(_, v)| self.matches_filter(v, filter));
109        }
110
111        // Apply sort
112        if let Some(ref sort) = self.pipeline.sort {
113            entities.sort_by(|(_, a), (_, b)| {
114                let a_val = extract_field(a, &sort.field_path);
115                let b_val = extract_field(b, &sort.field_path);
116                let cmp = compare_values(&a_val, &b_val);
117                match sort.order {
118                    SortOrder::Asc => cmp,
119                    SortOrder::Desc => cmp.reverse(),
120                }
121            });
122        }
123
124        // Apply limit
125        if let Some(limit) = self.pipeline.limit {
126            entities.truncate(limit);
127        }
128
129        // Update current keys
130        let keys: HashSet<String> = entities.iter().map(|(k, _)| k.clone()).collect();
131        *self.current_keys.write().await = keys;
132
133        entities
134    }
135
136    /// Check if an entity matches the filter
137    fn matches_filter(&self, entity: &Value, filter: &FilterConfig) -> bool {
138        let field_val = extract_field(entity, &filter.field_path);
139        match filter.op {
140            CompareOp::Eq => field_val == filter.value,
141            CompareOp::Ne => field_val != filter.value,
142            CompareOp::Gt => {
143                compare_values(&field_val, &filter.value) == std::cmp::Ordering::Greater
144            }
145            CompareOp::Gte => compare_values(&field_val, &filter.value) != std::cmp::Ordering::Less,
146            CompareOp::Lt => compare_values(&field_val, &filter.value) == std::cmp::Ordering::Less,
147            CompareOp::Lte => {
148                compare_values(&field_val, &filter.value) != std::cmp::Ordering::Greater
149            }
150        }
151    }
152
153    /// Determine the effect of an entity update on this view
154    pub async fn compute_effect(
155        &self,
156        key: &str,
157        new_value: Option<&Value>,
158        _cache: &EntityCache,
159    ) -> ViewEffect {
160        let current_keys = self.current_keys.read().await;
161        let was_in_view = current_keys.contains(key);
162        drop(current_keys);
163
164        // Check if entity now matches filter
165        let matches_now = match new_value {
166            Some(v) => {
167                if let Some(ref filter) = self.pipeline.filter {
168                    self.matches_filter(v, filter)
169                } else {
170                    true
171                }
172            }
173            None => false, // Deleted
174        };
175
176        match (was_in_view, matches_now) {
177            (false, true) => {
178                if self.pipeline.limit == Some(1) {
179                    let current_keys = self.current_keys.read().await;
180                    if let Some(current_key) = current_keys.iter().next() {
181                        if current_key != key {
182                            return ViewEffect::Replace {
183                                old_key: current_key.clone(),
184                                new_key: key.to_string(),
185                            };
186                        }
187                    }
188                }
189                ViewEffect::Add {
190                    key: key.to_string(),
191                }
192            }
193            (true, false) => ViewEffect::Remove {
194                key: key.to_string(),
195            },
196            (true, true) => ViewEffect::Update {
197                key: key.to_string(),
198            },
199            (false, false) => ViewEffect::NoEffect,
200        }
201    }
202
203    /// Apply an effect to update the current keys
204    pub async fn apply_effect(&self, effect: &ViewEffect) {
205        let mut keys = self.current_keys.write().await;
206        match effect {
207            ViewEffect::Add { key } => {
208                keys.insert(key.clone());
209            }
210            ViewEffect::Remove { key } => {
211                keys.remove(key);
212            }
213            ViewEffect::Replace { old_key, new_key } => {
214                keys.remove(old_key);
215                keys.insert(new_key.clone());
216            }
217            ViewEffect::Update { .. } | ViewEffect::NoEffect => {}
218        }
219    }
220}
221
222/// Extract a field value from a JSON object using a path
223fn extract_field(value: &Value, path: &[String]) -> Value {
224    let mut current = value;
225    for segment in path {
226        match current.get(segment) {
227            Some(v) => current = v,
228            None => return Value::Null,
229        }
230    }
231    current.clone()
232}
233
234/// Compare two JSON values
235fn compare_values(a: &Value, b: &Value) -> std::cmp::Ordering {
236    match (a, b) {
237        (Value::Number(a), Value::Number(b)) => {
238            let a_f = a.as_f64().unwrap_or(0.0);
239            let b_f = b.as_f64().unwrap_or(0.0);
240            a_f.partial_cmp(&b_f).unwrap_or(std::cmp::Ordering::Equal)
241        }
242        (Value::String(a), Value::String(b)) => a.cmp(b),
243        (Value::Bool(a), Value::Bool(b)) => a.cmp(b),
244        _ => std::cmp::Ordering::Equal,
245    }
246}
247
248/// Registry of materialized views
249#[derive(Default)]
250pub struct MaterializedViewRegistry {
251    views: HashMap<String, Arc<MaterializedView>>,
252    /// Map from source view ID to dependent materialized views
253    dependencies: HashMap<String, Vec<String>>,
254}
255
256impl MaterializedViewRegistry {
257    pub fn new() -> Self {
258        Self::default()
259    }
260
261    /// Register a materialized view
262    pub fn register(&mut self, view: MaterializedView) {
263        let view_id = view.id.clone();
264        let source_id = view.source_id.clone();
265
266        self.dependencies
267            .entry(source_id)
268            .or_default()
269            .push(view_id.clone());
270
271        self.views.insert(view_id, Arc::new(view));
272    }
273
274    /// Get a materialized view by ID
275    pub fn get(&self, id: &str) -> Option<Arc<MaterializedView>> {
276        self.views.get(id).cloned()
277    }
278
279    /// Get all views that depend on a source
280    pub fn get_dependents(&self, source_id: &str) -> Vec<Arc<MaterializedView>> {
281        self.dependencies
282            .get(source_id)
283            .map(|ids| {
284                ids.iter()
285                    .filter_map(|id| self.views.get(id).cloned())
286                    .collect()
287            })
288            .unwrap_or_default()
289    }
290}
291
292#[cfg(test)]
293mod tests {
294    use super::*;
295    use serde_json::json;
296
297    #[tokio::test]
298    async fn test_filter_evaluation() {
299        let pipeline = ViewPipeline {
300            filter: Some(FilterConfig {
301                field_path: vec!["status".to_string()],
302                op: CompareOp::Eq,
303                value: json!("active"),
304            }),
305            sort: None,
306            limit: None,
307        };
308
309        let view =
310            MaterializedView::new("test/active".to_string(), "test/list".to_string(), pipeline);
311
312        let entities = vec![
313            ("1".to_string(), json!({"status": "active", "value": 10})),
314            ("2".to_string(), json!({"status": "inactive", "value": 20})),
315            ("3".to_string(), json!({"status": "active", "value": 30})),
316        ];
317
318        let result = view.evaluate_pipeline(entities).await;
319        assert_eq!(result.len(), 2);
320        assert_eq!(result[0].0, "1");
321        assert_eq!(result[1].0, "3");
322    }
323
324    #[tokio::test]
325    async fn test_sort_and_limit() {
326        let pipeline = ViewPipeline {
327            filter: None,
328            sort: Some(SortConfig {
329                field_path: vec!["value".to_string()],
330                order: SortOrder::Desc,
331            }),
332            limit: Some(2),
333        };
334
335        let view =
336            MaterializedView::new("test/top2".to_string(), "test/list".to_string(), pipeline);
337
338        let entities = vec![
339            ("1".to_string(), json!({"value": 10})),
340            ("2".to_string(), json!({"value": 30})),
341            ("3".to_string(), json!({"value": 20})),
342        ];
343
344        let result = view.evaluate_pipeline(entities).await;
345        assert_eq!(result.len(), 2);
346        assert_eq!(result[0].0, "2"); // value: 30
347        assert_eq!(result[1].0, "3"); // value: 20
348    }
349}
arete_server/materialized_view.rs

arete_server/
materialized_view.rs
