rust_ef/db_context/
save_phases.rs

1//! Save pipeline phase functions, cascade drain helpers, and result type.
2//!
3//! - `drain_cascade_adds` / `drain_cascade_deletes`: extracted from
4//!   `save_changes` to keep `save_pipeline.rs` under 500 lines
5//! - 4 phase functions: per-DbSet DML execution, joining DbSet entries
6//!   with ChangeTracker state by `entry_id`
7//! - `SaveChangesResult`: save outcome summary
8
9use crate::change_executor::ChangeExecutor;
10use crate::db_set::DbSet;
11use crate::entity::{EntityState, IEntitySnapshot, IEntityType, IGetKeyValues};
12use crate::entity_snapshot::EntitySnapshot;
13use crate::error::{EFError, EFResult};
14use crate::metadata::EntityTypeMeta;
15use crate::provider::{IAsyncConnection, IDatabaseProvider};
16use crate::tracking::ChangeTracker;
17use std::any::TypeId;
18use std::collections::HashMap;
19
20use crate::cascade::{CascadeDeleteDirective, DrainedChild, FixupLink};
21
22// ---------------------------------------------------------------------------
23// Cascade drain helpers (extracted from save_changes)
24// ---------------------------------------------------------------------------
25
26impl super::DbContext {
27    /// Iteratively drains HasMany/M2M children from Added principals.
28    pub(super) fn drain_cascade_adds(
29        &mut self,
30        type_ids: &[TypeId],
31        configured_metas: &HashMap<TypeId, EntityTypeMeta>,
32    ) -> EFResult<Vec<FixupLink>> {
33        let mut fixup_links: Vec<FixupLink> = Vec::new();
34        loop {
35            let mut all_drained: Vec<DrainedChild> = Vec::new();
36            for type_id in type_ids {
37                let Some(saver) = self.savers.get(type_id) else {
38                    continue;
39                };
40                let Some(set) = self.sets.get_mut(type_id) else {
41                    continue;
42                };
43                let meta = configured_metas
44                    .get(type_id)
45                    .or_else(|| self.entity_metas.get(type_id))
46                    .ok_or_else(|| {
47                        EFError::configuration(format!(
48                            "entity metadata not found for {:?}",
49                            type_id
50                        ))
51                    })?;
52                all_drained.extend(saver.drain_cascade_children(
53                    set.as_mut(),
54                    &self.change_tracker,
55                    meta,
56                ));
57            }
58            if all_drained.is_empty() {
59                break;
60            }
61            for child in all_drained {
62                let child_saver = self.savers.get(&child.child_type_id).ok_or_else(|| {
63                    EFError::configuration(format!(
64                        "Cannot cascade-save child type {:?}: no DbSet registered. \
65                         Call ctx.set::<ChildType>() before save_changes.",
66                        child.child_type_id
67                    ))
68                })?;
69                let child_set = self.sets.get_mut(&child.child_type_id).ok_or_else(|| {
70                    EFError::configuration(format!(
71                        "DbSet not found for registered saver type {:?}",
72                        child.child_type_id
73                    ))
74                })?;
75                if let Some(child_idx) = child_saver.add_cascade_child(
76                    child_set.as_mut(),
77                    &mut self.change_tracker,
78                    child.child,
79                ) {
80                    if let Some(link) = fixup_links.iter_mut().find(|l| {
81                        l.parent_type_id == child.parent_type_id
82                            && l.parent_entry_idx == child.parent_entry_idx
83                            && l.child_type_id == child.child_type_id
84                            && l.through_table == child.through_table
85                    }) {
86                        link.child_entry_indices.push(child_idx);
87                    } else {
88                        fixup_links.push(FixupLink {
89                            parent_type_id: child.parent_type_id,
90                            parent_entry_idx: child.parent_entry_idx,
91                            child_type_id: child.child_type_id,
92                            child_entry_indices: vec![child_idx],
93                            fk_target_type_id: child.fk_target_type_id,
94                            through_table: child.through_table,
95                            through_parent_fk_col: child.through_parent_fk_col,
96                            through_child_fk_col: child.through_child_fk_col,
97                        });
98                    }
99                }
100            }
101        }
102        Ok(fixup_links)
103    }
104
105    /// Iteratively drains HasMany children from Deleted principals and collects
106    /// direct DELETE/SET NULL directives for untracked dependents.
107    pub(super) fn drain_cascade_deletes(
108        &mut self,
109        type_ids: &[TypeId],
110        configured_metas: &HashMap<TypeId, EntityTypeMeta>,
111        processed: &mut std::collections::HashSet<(TypeId, usize)>,
112    ) -> EFResult<Vec<CascadeDeleteDirective>> {
113        let mut delete_directives: Vec<CascadeDeleteDirective> = Vec::new();
114        loop {
115            let mut all_drained_deleted: Vec<DrainedChild> = Vec::new();
116            for type_id in type_ids {
117                let Some(saver) = self.savers.get(type_id) else {
118                    continue;
119                };
120                let Some(set) = self.sets.get_mut(type_id) else {
121                    continue;
122                };
123                let meta = configured_metas
124                    .get(type_id)
125                    .or_else(|| self.entity_metas.get(type_id))
126                    .ok_or_else(|| {
127                        EFError::configuration(format!(
128                            "entity metadata not found for {:?}",
129                            type_id
130                        ))
131                    })?;
132                let (drained, directives) = saver.drain_cascade_deleted_children(
133                    set.as_mut(),
134                    &self.change_tracker,
135                    meta,
136                    processed,
137                );
138                all_drained_deleted.extend(drained);
139                delete_directives.extend(directives);
140            }
141            if all_drained_deleted.is_empty() {
142                break;
143            }
144            for child in all_drained_deleted {
145                let child_saver = self.savers.get(&child.child_type_id).ok_or_else(|| {
146                    EFError::configuration(format!(
147                        "Cannot cascade-delete child type {:?}: no DbSet registered. \
148                         Call ctx.set::<ChildType>() before save_changes.",
149                        child.child_type_id
150                    ))
151                })?;
152                let child_set = self.sets.get_mut(&child.child_type_id).ok_or_else(|| {
153                    EFError::configuration(format!(
154                        "DbSet not found for registered saver type {:?}",
155                        child.child_type_id
156                    ))
157                })?;
158                child_saver.add_cascade_deleted_child(
159                    child_set.as_mut(),
160                    &mut self.change_tracker,
161                    child.child,
162                );
163            }
164        }
165        Ok(delete_directives)
166    }
167}
168
169// ---------------------------------------------------------------------------
170// Per-DbSet phase functions — join DbSet entries with ChangeTracker by entry_id
171// ---------------------------------------------------------------------------
172
173/// Phase 1a: INSERT Added (non-upsert) entities, then backfill generated PKs.
174pub async fn insert_added_phase<E>(
175    conn: &mut dyn IAsyncConnection,
176    provider: &dyn IDatabaseProvider,
177    db_set: &mut DbSet<E>,
178    tracker: &ChangeTracker,
179    meta: &EntityTypeMeta,
180) -> EFResult<usize>
181where
182    E: IEntityType + IEntitySnapshot + IGetKeyValues,
183{
184    let added_indices: Vec<usize> = db_set
185        .entries
186        .iter()
187        .enumerate()
188        .filter(|(_, e)| tracker.entry_state(e.entry_id) == Some(EntityState::Added))
189        .filter(|(_, e)| !tracker.entry_is_upsert(e.entry_id))
190        .map(|(i, _)| i)
191        .collect();
192    if added_indices.is_empty() {
193        return Ok(0);
194    }
195    let added_refs: Vec<(&E, &EntityTypeMeta)> = added_indices
196        .iter()
197        .map(|&i| (&db_set.entries[i].entity, meta))
198        .collect();
199    let mut generated_keys: Vec<i64> = vec![0; added_indices.len()];
200    let inserted = ChangeExecutor::execute_inserts(conn, provider, &added_refs, |idx, key| {
201        if idx < generated_keys.len() {
202            generated_keys[idx] = key;
203        }
204    })
205    .await?;
206    for (i, &entry_idx) in added_indices.iter().enumerate() {
207        if let Some(&key) = generated_keys.get(i) {
208            if key != 0 {
209                db_set.entries[entry_idx].entity.set_auto_increment_key(key);
210            }
211        }
212    }
213    Ok(inserted)
214}
215
216/// Phase 1b: UPSERT Added entities (is_upsert = true).
217pub async fn upsert_added_phase<E>(
218    conn: &mut dyn IAsyncConnection,
219    provider: &dyn IDatabaseProvider,
220    db_set: &DbSet<E>,
221    tracker: &ChangeTracker,
222    meta: &EntityTypeMeta,
223) -> EFResult<usize>
224where
225    E: IEntityType + IEntitySnapshot + IGetKeyValues,
226{
227    let upserts: Vec<(&E, &EntityTypeMeta)> = db_set
228        .entries
229        .iter()
230        .filter(|e| tracker.entry_state(e.entry_id) == Some(EntityState::Added))
231        .filter(|e| tracker.entry_is_upsert(e.entry_id))
232        .map(|e| (&e.entity, meta))
233        .collect();
234    if upserts.is_empty() {
235        return Ok(0);
236    }
237    ChangeExecutor::execute_upserts(conn, provider, &upserts).await
238}
239
240/// Phase 2: UPDATE Modified entities (partial update via modified_properties).
241pub async fn update_modified_phase<E>(
242    conn: &mut dyn IAsyncConnection,
243    provider: &dyn IDatabaseProvider,
244    db_set: &DbSet<E>,
245    tracker: &ChangeTracker,
246    meta: &EntityTypeMeta,
247    query_filter: Option<&crate::query::BoolExpr>,
248) -> EFResult<usize>
249where
250    E: IEntityType + IEntitySnapshot + IGetKeyValues,
251{
252    let modified: Vec<(&E, &EntityTypeMeta, Option<&EntitySnapshot>, &[String])> = db_set
253        .entries
254        .iter()
255        .filter(|e| tracker.entry_state(e.entry_id) == Some(EntityState::Modified))
256        .map(|e| {
257            let original = tracker.entry_original(e.entry_id);
258            let modified_props = tracker.entry_modified(e.entry_id).unwrap_or(&[]);
259            (&e.entity, meta, original, modified_props)
260        })
261        .collect();
262    if modified.is_empty() {
263        return Ok(0);
264    }
265    ChangeExecutor::execute_updates(conn, provider, &modified, query_filter).await
266}
267
268/// Phase 3: DELETE Deleted entities.
269pub async fn delete_deleted_phase<E>(
270    conn: &mut dyn IAsyncConnection,
271    provider: &dyn IDatabaseProvider,
272    db_set: &DbSet<E>,
273    tracker: &ChangeTracker,
274    meta: &EntityTypeMeta,
275    query_filter: Option<&crate::query::BoolExpr>,
276) -> EFResult<usize>
277where
278    E: IEntityType + IEntitySnapshot + IGetKeyValues,
279{
280    let deleted: Vec<(&E, &EntityTypeMeta, Option<&EntitySnapshot>)> = db_set
281        .entries
282        .iter()
283        .filter(|e| tracker.entry_state(e.entry_id) == Some(EntityState::Deleted))
284        .map(|e| {
285            let original = tracker.entry_original(e.entry_id);
286            (&e.entity, meta, original)
287        })
288        .collect();
289    if deleted.is_empty() {
290        return Ok(0);
291    }
292    ChangeExecutor::execute_deletes(conn, provider, &deleted, query_filter).await
293}
294
295// ---------------------------------------------------------------------------
296// SaveChangesResult
297// ---------------------------------------------------------------------------
298
299#[derive(Debug, Clone)]
300pub struct SaveChangesResult {
301    pub added: usize,
302    pub updated: usize,
303    pub deleted: usize,
304}
305impl SaveChangesResult {
306    pub fn total(&self) -> usize {
307        self.added + self.updated + self.deleted
308    }
309}
310impl std::fmt::Display for SaveChangesResult {
311    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
312        write!(
313            f,
314            "{} entities modified ({} added, {} updated, {} deleted)",
315            self.total(),
316            self.added,
317            self.updated,
318            self.deleted
319        )
320    }
321}
rust_ef/db_context/save_phases.rs

rust_ef/db_context/
save_phases.rs
