arete_server/view/
registry.rs

1use crate::sorted_cache::{SortOrder, SortedViewCache};
2use crate::view::ViewSpec;
3use std::collections::HashMap;
4use std::sync::Arc;
5use tokio::sync::RwLock;
6
7#[derive(Clone)]
8pub struct ViewIndex {
9    by_export: HashMap<String, Vec<ViewSpec>>,
10    by_id: HashMap<String, ViewSpec>,
11    sorted_caches: Arc<RwLock<HashMap<String, SortedViewCache>>>,
12    /// Map from source view ID to derived view IDs
13    derived_by_source: HashMap<String, Vec<String>>,
14}
15
16impl ViewIndex {
17    pub fn new() -> Self {
18        Self {
19            by_export: HashMap::new(),
20            by_id: HashMap::new(),
21            sorted_caches: Arc::new(RwLock::new(HashMap::new())),
22            derived_by_source: HashMap::new(),
23        }
24    }
25
26    pub fn add_spec(&mut self, spec: ViewSpec) {
27        if let Some(ref source) = spec.source_view {
28            self.derived_by_source
29                .entry(source.clone())
30                .or_default()
31                .push(spec.id.clone());
32        }
33
34        if let Some(ref pipeline) = spec.pipeline {
35            if let Some(ref sort_config) = pipeline.sort {
36                self.init_sorted_cache_sync(
37                    &spec.id,
38                    sort_config.field_path.clone(),
39                    sort_config.order.into(),
40                );
41            }
42        }
43
44        // Only add non-derived views to by_export.
45        // Derived views receive updates via their source_view subscription,
46        // not directly from the projector.
47        if !spec.is_derived() {
48            self.by_export
49                .entry(spec.export.clone())
50                .or_default()
51                .push(spec.clone());
52        }
53        self.by_id.insert(spec.id.clone(), spec);
54    }
55
56    pub fn by_export(&self, entity: &str) -> &[ViewSpec] {
57        self.by_export
58            .get(entity)
59            .map(|v| v.as_slice())
60            .unwrap_or(&[])
61    }
62
63    pub fn get_view(&self, id: &str) -> Option<&ViewSpec> {
64        self.by_id.get(id)
65    }
66
67    /// Stable view order for snapshot compatibility contracts.
68    pub(crate) fn snapshot_specs(&self) -> Vec<&ViewSpec> {
69        let mut specs = self.by_id.values().collect::<Vec<_>>();
70        specs.sort_by(|left, right| left.id.cmp(&right.id));
71        specs
72    }
73
74    pub fn get_derived_views(&self) -> Vec<&ViewSpec> {
75        self.by_id.values().filter(|s| s.is_derived()).collect()
76    }
77
78    pub fn get_derived_views_for_source(&self, source_view_id: &str) -> Vec<&ViewSpec> {
79        self.derived_by_source
80            .get(source_view_id)
81            .map(|ids| ids.iter().filter_map(|id| self.by_id.get(id)).collect())
82            .unwrap_or_default()
83    }
84
85    pub fn sorted_caches(&self) -> Arc<RwLock<HashMap<String, SortedViewCache>>> {
86        self.sorted_caches.clone()
87    }
88
89    pub async fn init_sorted_cache(
90        &self,
91        view_id: &str,
92        sort_field: Vec<String>,
93        order: SortOrder,
94    ) {
95        let mut caches = self.sorted_caches.write().await;
96        if !caches.contains_key(view_id) {
97            caches.insert(
98                view_id.to_string(),
99                SortedViewCache::new(view_id.to_string(), sort_field, order),
100            );
101        }
102    }
103
104    fn init_sorted_cache_sync(&mut self, view_id: &str, sort_field: Vec<String>, order: SortOrder) {
105        let cache = SortedViewCache::new(view_id.to_string(), sort_field, order);
106        let caches = Arc::get_mut(&mut self.sorted_caches)
107            .expect("Cannot initialize sorted cache: Arc is shared");
108        let caches = caches.get_mut();
109        if !caches.contains_key(view_id) {
110            caches.insert(view_id.to_string(), cache);
111        }
112    }
113}
114
115impl Default for ViewIndex {
116    fn default() -> Self {
117        Self::new()
118    }
119}
arete_server/view/registry.rs

arete_server/view/
registry.rs
