icydb_core/db/executor/
save.rs

1use crate::{
2    Error,
3    db::{
4        Db,
5        executor::ExecutorError,
6        query::{SaveMode, SaveQuery},
7        store::DataKey,
8    },
9    deserialize,
10    obs::metrics,
11    serialize,
12    traits::EntityKind,
13    visitor::{sanitize, validate},
14};
15use std::marker::PhantomData;
16
17///
18/// SaveExecutor
19///
20
21#[derive(Clone, Copy)]
22pub struct SaveExecutor<E: EntityKind> {
23    db: Db<E::Canister>,
24    debug: bool,
25    _marker: PhantomData<E>,
26}
27
28impl<E: EntityKind> SaveExecutor<E> {
29    #[must_use]
30    pub const fn new(db: Db<E::Canister>, debug: bool) -> Self {
31        Self {
32            db,
33            debug,
34            _marker: PhantomData,
35        }
36    }
37
38    // debug
39    #[must_use]
40    pub const fn debug(mut self) -> Self {
41        self.debug = true;
42        self
43    }
44
45    ///
46    /// EXECUTION METHODS
47    ///
48
49    pub fn insert(&self, entity: E) -> Result<E, Error> {
50        let entity = self.save_entity(SaveMode::Insert, entity)?;
51
52        Ok(entity)
53    }
54
55    pub fn insert_view<V>(&self, view: E::ViewType) -> Result<E::ViewType, Error> {
56        let entity = E::from_view(view);
57        let saved_view = self.insert(entity)?.to_view();
58
59        Ok(saved_view)
60    }
61
62    pub fn update(&self, entity: E) -> Result<E, Error> {
63        let entity = self.save_entity(SaveMode::Update, entity)?;
64
65        Ok(entity)
66    }
67
68    pub fn update_view<V>(&self, view: E::ViewType) -> Result<E::ViewType, Error> {
69        let entity = E::from_view(view);
70        let saved_view = self.update(entity)?.to_view();
71
72        Ok(saved_view)
73    }
74
75    pub fn replace(&self, entity: E) -> Result<E, Error> {
76        let entity = self.save_entity(SaveMode::Replace, entity)?;
77
78        Ok(entity)
79    }
80
81    pub fn replace_view<V>(&self, view: E::ViewType) -> Result<E::ViewType, Error> {
82        let entity = E::from_view(view);
83        let saved_view = self.replace(entity)?.to_view();
84
85        Ok(saved_view)
86    }
87
88    // execute
89    // serializes the save query to pass to save_entity
90    pub fn execute(&self, query: SaveQuery) -> Result<E, Error> {
91        let e: E = deserialize(&query.bytes)?;
92        let entity = self.save_entity(query.mode, e)?;
93
94        Ok(entity)
95    }
96
97    // save_entity
98    fn save_entity(&self, mode: SaveMode, mut entity: E) -> Result<E, Error> {
99        let mut span = metrics::Span::<E>::new(metrics::ExecKind::Save);
100        let key = entity.key();
101        let ctx = self.db.context::<E>();
102
103        // sanitize & validate
104        sanitize(&mut entity);
105        validate(&entity)?;
106
107        // debug
108        //   debug!(self.debug, "query.{mode}: {} ({key:?}) ", E::PATH);
109
110        //
111        // match save mode
112        // on Update and Replace compare old and new data
113        //
114
115        let data_key = DataKey::new::<E>(key);
116        let old_result = ctx.with_store(|store| store.get(&data_key))?;
117
118        // did anything change?
119        let old = match (mode, old_result) {
120            (SaveMode::Insert | SaveMode::Replace, None) => None,
121
122            (SaveMode::Update | SaveMode::Replace, Some(old_bytes)) => {
123                let old = deserialize::<E>(&old_bytes)?;
124                Some(old)
125            }
126
127            // invalid
128            (SaveMode::Insert, Some(_)) => return Err(ExecutorError::KeyExists(data_key))?,
129            (SaveMode::Update, None) => return Err(ExecutorError::KeyNotFound(data_key))?,
130        };
131
132        // now we can serialize
133        let bytes = serialize(&entity)?;
134
135        // replace indexes, fail if there are any unique violations
136        self.replace_indexes(old.as_ref(), &entity)?;
137
138        // insert data row
139        ctx.with_store_mut(|store| store.insert(data_key.clone(), bytes))?;
140        span.set_rows(1);
141
142        Ok(entity)
143    }
144
145    // replace_indexes: two-phase (validate, then mutate) to avoid partial updates
146    fn replace_indexes(&self, old: Option<&E>, new: &E) -> Result<(), Error> {
147        use crate::db::store::IndexKey;
148
149        // Phase 1: validate uniqueness for all indexes without mutating
150        for index in E::INDEXES {
151            // Only check when we can compute the new key and the index is unique
152            if index.unique
153                && let Some(new_idx_key) = IndexKey::new(new, index)
154            {
155                let store = self.db.with_index(|reg| reg.try_get_store(index.store))?;
156                let violates = store.with_borrow(|s| {
157                    if let Some(existing) = s.get(&new_idx_key) {
158                        let new_entity_key = new.key();
159                        !existing.contains(&new_entity_key) && !existing.is_empty()
160                    } else {
161                        false
162                    }
163                });
164                if violates {
165                    // Count the unique violation just like the store-level check would have
166                    metrics::with_state_mut(|m| {
167                        metrics::record_unique_violation_for::<E>(m);
168                    });
169
170                    return Err(ExecutorError::index_violation(E::PATH, index.fields).into());
171                }
172            }
173        }
174
175        // Phase 2: apply changes (remove old, insert new) for each index
176        for index in E::INDEXES {
177            let store = self.db.with_index(|reg| reg.try_get_store(index.store))?;
178            store.with_borrow_mut(|s| {
179                if let Some(old) = old {
180                    s.remove_index_entry(old, index);
181                }
182                s.insert_index_entry(new, index)?;
183
184                Ok::<(), Error>(())
185            })?;
186        }
187
188        Ok(())
189    }
190}
icydb_core/db/executor/save.rs

icydb_core/db/executor/
save.rs
