lref/
change_executor.rs

1//! Change executor — generates and executes SQL for entity state changes.
2//!
3//! The `ChangeExecutor` takes a collection of tracked entities grouped by
4//! state (Added/Modified/Deleted), generates the appropriate parameterized
5//! DML, and executes it against the database via the provider.
6
7use crate::entity::{IEntitySnapshot, IEntityType, IGetKeyValues};
8use crate::error::LrefResult;
9use crate::metadata::EntityTypeMeta;
10use crate::provider::{DbValue, IAsyncConnection, IDatabaseProvider};
11use std::collections::HashMap;
12
13/// Executes INSERT/UPDATE/DELETE for tracked entities within a transaction.
14pub struct ChangeExecutor;
15
16impl ChangeExecutor {
17    /// Executes INSERT statements for all added entities.
18    /// Returns the number of rows inserted.
19    /// For auto-increment columns, the generated key values are written back
20    /// via the `on_key_backfill` callback.
21    pub async fn execute_inserts<E, F>(
22        conn: &mut dyn IAsyncConnection,
23        provider: &dyn IDatabaseProvider,
24        entities: &[(&E, &EntityTypeMeta)],
25        mut on_key_backfill: F,
26    ) -> LrefResult<usize>
27    where
28        E: IEntityType + IEntitySnapshot + IGetKeyValues,
29        F: FnMut(usize, i64),
30    {
31        let gen = provider.sql_generator();
32        let mut inserted = 0;
33
34        for (idx, (entity, meta)) in entities.iter().enumerate() {
35            let snap = entity.snapshot();
36            let scalar_props: Vec<_> = meta.mapped_scalar_properties().collect();
37            if scalar_props.is_empty() {
38                continue;
39            }
40
41            // Build column list (exclude auto-increment primary keys for INSERT)
42            let insert_cols: Vec<&str> = scalar_props
43                .iter()
44                .filter(|p| !p.is_auto_increment || !p.is_primary_key)
45                .map(|p| p.column_name.as_ref())
46                .collect();
47
48            let params: Vec<DbValue> = scalar_props
49                .iter()
50                .filter(|p| !p.is_auto_increment || !p.is_primary_key)
51                .map(|p| {
52                    snap.get(p.field_name.as_ref())
53                        .cloned()
54                        .unwrap_or(DbValue::Null)
55                })
56                .collect();
57
58            if insert_cols.is_empty() {
59                continue;
60            }
61
62            let sql = gen.insert(meta.table_name.as_ref(), &insert_cols, true);
63            let rows = conn.execute(&sql, &params).await?;
64
65            // Backfill auto-generated key from RETURNING
66            if rows > 0 {
67                on_key_backfill(idx, rows as i64);
68                inserted += 1;
69            }
70        }
71
72        Ok(inserted)
73    }
74
75    /// Executes UPDATE statements for all modified entities.
76    pub async fn execute_updates<E>(
77        conn: &mut dyn IAsyncConnection,
78        provider: &dyn IDatabaseProvider,
79        entities: &[(&E, &EntityTypeMeta)],
80    ) -> LrefResult<usize>
81    where
82        E: IEntityType + IEntitySnapshot + IGetKeyValues,
83    {
84        let gen = provider.sql_generator();
85        let mut updated = 0;
86
87        for (_entity, meta) in entities {
88            let snap = _entity.snapshot();
89            let keys = _entity.key_values();
90            let scalar_props: Vec<_> = meta.mapped_scalar_properties().collect();
91
92            // SET columns = non-PK + non-auto-increment
93            let set_cols: Vec<&str> = scalar_props
94                .iter()
95                .filter(|p| !p.is_primary_key)
96                .map(|p| p.column_name.as_ref())
97                .collect();
98
99            if set_cols.is_empty() || keys.is_empty() {
100                continue;
101            }
102
103            // Build WHERE clause from primary key
104            let where_parts: Vec<String> = keys
105                .keys()
106                .enumerate()
107                .map(|(i, k)| {
108                    format!(
109                        "{} = {}",
110                        gen.quote_identifier(k),
111                        gen.parameter_placeholder(i + 1)
112                    )
113                })
114                .collect();
115            let where_clause = where_parts.join(" AND ");
116
117            let sql = gen.update(meta.table_name.as_ref(), &set_cols, &where_clause);
118
119            // Params: SET values first, then WHERE values
120            let mut params: Vec<DbValue> = set_cols
121                .iter()
122                .map(|col| {
123                    // Map column_name back to field_name
124                    let prop = scalar_props.iter().find(|p| p.column_name.as_ref() == *col);
125                    match prop {
126                        Some(p) => snap
127                            .get(p.field_name.as_ref())
128                            .cloned()
129                            .unwrap_or(DbValue::Null),
130                        None => DbValue::Null,
131                    }
132                })
133                .collect();
134
135            for (_k, v) in &keys {
136                params.push(v.clone());
137            }
138
139            let rows = conn.execute(&sql, &params).await?;
140            if rows > 0 {
141                updated += 1;
142            }
143        }
144
145        Ok(updated)
146    }
147
148    /// Executes DELETE statements for all deleted entities.
149    pub async fn execute_deletes<E>(
150        conn: &mut dyn IAsyncConnection,
151        provider: &dyn IDatabaseProvider,
152        entities: &[(&E, &EntityTypeMeta)],
153    ) -> LrefResult<usize>
154    where
155        E: IEntityType + IGetKeyValues,
156    {
157        let gen = provider.sql_generator();
158        let mut deleted = 0;
159
160        for (_entity, meta) in entities {
161            let keys = _entity.key_values();
162            if keys.is_empty() {
163                continue;
164            }
165
166            let where_parts: Vec<String> = keys
167                .keys()
168                .enumerate()
169                .map(|(i, k)| {
170                    format!(
171                        "{} = {}",
172                        gen.quote_identifier(k),
173                        gen.parameter_placeholder(i + 1)
174                    )
175                })
176                .collect();
177            let where_clause = where_parts.join(" AND ");
178
179            let sql = gen.delete(meta.table_name.as_ref(), &where_clause);
180
181            let params: Vec<DbValue> = keys.values().cloned().collect();
182            let rows = conn.execute(&sql, &params).await?;
183            if rows > 0 {
184                deleted += 1;
185            }
186        }
187
188        Ok(deleted)
189    }
190}
191
192// ---------------------------------------------------------------------------
193// Standalone SQL generation helpers (for use by simplified callers)
194// ---------------------------------------------------------------------------
195
196pub fn generate_insert_sql(
197    provider: &dyn IDatabaseProvider,
198    meta: &EntityTypeMeta,
199    _property_values: &HashMap<String, DbValue>,
200) -> String {
201    let gen = provider.sql_generator();
202    let scalar_props: Vec<_> = meta.mapped_scalar_properties().collect();
203    let columns: Vec<&str> = scalar_props
204        .iter()
205        .map(|p| p.column_name.as_ref())
206        .collect();
207    if columns.is_empty() {
208        return String::new();
209    }
210    gen.insert(meta.table_name.as_ref(), &columns, true)
211}
212
213pub fn generate_update_sql(
214    provider: &dyn IDatabaseProvider,
215    meta: &EntityTypeMeta,
216    property_values: &HashMap<String, DbValue>,
217    primary_key_values: &HashMap<String, DbValue>,
218) -> String {
219    let gen = provider.sql_generator();
220    let set_columns: Vec<&str> = property_values
221        .keys()
222        .filter(|k| !primary_key_values.contains_key(*k))
223        .map(|k| k.as_str())
224        .collect();
225    if set_columns.is_empty() || primary_key_values.is_empty() {
226        return String::new();
227    }
228    let where_parts: Vec<String> = primary_key_values
229        .keys()
230        .enumerate()
231        .map(|(i, k)| {
232            format!(
233                "{} = {}",
234                gen.quote_identifier(k),
235                gen.parameter_placeholder(i + 1)
236            )
237        })
238        .collect();
239    gen.update(
240        meta.table_name.as_ref(),
241        &set_columns,
242        &where_parts.join(" AND "),
243    )
244}
245
246pub fn generate_delete_sql(
247    provider: &dyn IDatabaseProvider,
248    meta: &EntityTypeMeta,
249    primary_key_values: &HashMap<String, DbValue>,
250) -> String {
251    let gen = provider.sql_generator();
252    if primary_key_values.is_empty() {
253        return String::new();
254    }
255    let where_parts: Vec<String> = primary_key_values
256        .keys()
257        .enumerate()
258        .map(|(i, k)| {
259            format!(
260                "{} = {}",
261                gen.quote_identifier(k),
262                gen.parameter_placeholder(i + 1)
263            )
264        })
265        .collect();
266    gen.delete(meta.table_name.as_ref(), &where_parts.join(" AND "))
267}
268
269pub fn collect_insert_params(
270    meta: &EntityTypeMeta,
271    property_values: &HashMap<String, DbValue>,
272) -> Vec<DbValue> {
273    meta.mapped_scalar_properties()
274        .map(|p| {
275            property_values
276                .get(p.field_name.as_ref())
277                .cloned()
278                .unwrap_or(DbValue::Null)
279        })
280        .collect()
281}
282
283pub fn collect_update_params(
284    property_values: &HashMap<String, DbValue>,
285    primary_key_values: &HashMap<String, DbValue>,
286    set_keys: &[String],
287) -> Vec<DbValue> {
288    let mut params: Vec<DbValue> = set_keys
289        .iter()
290        .filter(|k| !primary_key_values.contains_key(*k))
291        .map(|k| property_values.get(k).cloned().unwrap_or(DbValue::Null))
292        .collect();
293    for v in primary_key_values.values() {
294        params.push(v.clone());
295    }
296    params
297}
298
299pub fn collect_delete_params(primary_key_values: &HashMap<String, DbValue>) -> Vec<DbValue> {
300    primary_key_values.values().cloned().collect()
301}
lref/change_executor.rs

lref/
change_executor.rs
