rust_ef/db_context/
save_phases.rs

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

rust_ef/db_context/
save_phases.rs
