architect_sdk/service/
crud.rs

1//! Generic CRUD execution against PostgreSQL.
2
3use crate::config::{IncludeSpec, ResolvedEntity};
4use crate::db::pool::{Connection, DbRow, Pool};
5use crate::db::Dialect;
6use crate::error::AppError;
7use crate::extensible_fields::ExtensibleRegistry;
8use crate::sql::{
9    archive, coerce_json_value_for_pg_array, delete, insert, insert_history_snapshot,
10    prune_history, select_by_column_in, select_by_id, select_list, select_list_with_includes,
11    unarchive, update, BindValue, FilterNode, IncludeSelect, QueryBuf, SortSpec,
12};
13use serde_json::Value;
14use std::collections::HashMap;
15
16/// Execution target: either a pool (for database/schema strategy) or a single connection (for RLS, with SET LOCAL already applied).
17pub enum TenantExecutorInner<'a> {
18    Pool(&'a Pool),
19    Conn(&'a mut Connection),
20}
21
22pub struct TenantExecutor<'a> {
23    pub executor: TenantExecutorInner<'a>,
24    pub dialect: &'a dyn crate::db::Dialect,
25}
26
27impl<'a> TenantExecutor<'a> {
28    pub fn pool(pool: &'a Pool, dialect: &'a dyn crate::db::Dialect) -> Self {
29        TenantExecutor {
30            executor: TenantExecutorInner::Pool(pool),
31            dialect,
32        }
33    }
34    pub fn conn(conn: &'a mut Connection, dialect: &'a dyn crate::db::Dialect) -> Self {
35        TenantExecutor {
36            executor: TenantExecutorInner::Conn(conn),
37            dialect,
38        }
39    }
40}
41
42/// A child group for [`CrudService::create_graph`]: the to-many include spec, the resolved
43/// child entity, and the list of child bodies to insert under it.
44pub type GraphChild = (IncludeSpec, ResolvedEntity, Vec<HashMap<String, Value>>);
45
46pub struct CrudService;
47
48impl CrudService {
49    /// List rows with optional RSQL filter and sort, limit (default 100, max 1000), offset (default 0).
50    /// `filter_includes` supplies related-entity metadata for dotted-field EXISTS filters; pass `&[]` when unused.
51    #[allow(clippy::too_many_arguments)]
52    pub async fn list<'a>(
53        executor: &mut TenantExecutor<'a>,
54        entity: &ResolvedEntity,
55        filter: Option<&FilterNode>,
56        sort: &[SortSpec],
57        limit: Option<u32>,
58        offset: Option<u32>,
59        filter_includes: &[IncludeSelect<'_>],
60        schema_override: Option<&str>,
61        dialect: &dyn Dialect,
62        registry: Option<&ExtensibleRegistry>,
63    ) -> Result<Vec<Value>, AppError> {
64        const DEFAULT_LIMIT: u32 = 100;
65        let limit = limit.unwrap_or(DEFAULT_LIMIT).min(1000);
66        let offset = offset.unwrap_or(0);
67        let q = select_list(
68            entity,
69            filter,
70            sort,
71            Some(limit),
72            Some(offset),
73            filter_includes,
74            schema_override,
75            dialect,
76            registry,
77        )?;
78        Self::query_many_exec(executor, &q.sql, &q.params).await
79    }
80
81    /// List rows with includes in a single query (scalar subqueries with json_agg/row_to_json). Returns rows with include keys already set (JSON).
82    /// `includes` drives scalar subqueries for response data; `filter_includes` is the superset used for EXISTS generation.
83    #[allow(clippy::too_many_arguments)]
84    pub async fn list_with_includes<'a>(
85        executor: &mut TenantExecutor<'a>,
86        entity: &ResolvedEntity,
87        filter: Option<&FilterNode>,
88        sort: &[SortSpec],
89        limit: Option<u32>,
90        offset: Option<u32>,
91        includes: &[IncludeSelect<'_>],
92        filter_includes: &[IncludeSelect<'_>],
93        schema_override: Option<&str>,
94        dialect: &dyn Dialect,
95        registry: Option<&ExtensibleRegistry>,
96    ) -> Result<Vec<Value>, AppError> {
97        const DEFAULT_LIMIT: u32 = 100;
98        let limit = limit.unwrap_or(DEFAULT_LIMIT).min(1000);
99        let offset = offset.unwrap_or(0);
100        let q = select_list_with_includes(
101            entity,
102            filter,
103            sort,
104            Some(limit),
105            Some(offset),
106            includes,
107            filter_includes,
108            schema_override,
109            dialect,
110            registry,
111        )?;
112        Self::query_many_exec(executor, &q.sql, &q.params).await
113    }
114
115    /// Fetch one row by primary key. Returns JSON object or None.
116    pub async fn read<'a>(
117        executor: &mut TenantExecutor<'a>,
118        entity: &ResolvedEntity,
119        id: &Value,
120        schema_override: Option<&str>,
121        dialect: &dyn Dialect,
122    ) -> Result<Option<Value>, AppError> {
123        let q = select_by_id(entity, schema_override, dialect);
124        Self::query_one_exec(executor, &q.sql, std::slice::from_ref(id)).await
125    }
126
127    /// Fetch rows from entity where column IN (values). Used for batch-loading related rows.
128    pub async fn fetch_where_column_in<'a>(
129        executor: &mut TenantExecutor<'a>,
130        entity: &ResolvedEntity,
131        column_name: &str,
132        values: &[Value],
133        schema_override: Option<&str>,
134        dialect: &dyn Dialect,
135    ) -> Result<Vec<Value>, AppError> {
136        if values.is_empty() {
137            return Ok(Vec::new());
138        }
139        let q = select_by_column_in(entity, column_name, values, schema_override, dialect);
140        Self::query_many_exec(executor, &q.sql, &q.params).await
141    }
142
143    /// Insert one row; body may include or omit PK (if has default). Returns created row.
144    /// When rls_tenant_id is Some (RLS strategy), tenant_id column is set automatically.
145    /// When caller_user_id is Some, created_by is set to that value.
146    pub async fn create<'a>(
147        executor: &mut TenantExecutor<'a>,
148        entity: &ResolvedEntity,
149        body: &HashMap<String, Value>,
150        schema_override: Option<&str>,
151        rls_tenant_id: Option<&str>,
152        caller_user_id: Option<&str>,
153        dialect: &dyn Dialect,
154    ) -> Result<Value, AppError> {
155        let include_pk = body.contains_key(&entity.pk_columns[0]);
156        let q = insert(
157            entity,
158            body,
159            include_pk,
160            schema_override,
161            rls_tenant_id,
162            caller_user_id,
163            dialect,
164        );
165        let row = Self::execute_returning_one_exec(executor, &q)
166            .await?
167            .ok_or_else(|| AppError::Db(sqlx::Error::RowNotFound))?;
168        if entity.audit_log {
169            Self::insert_audit(
170                executor,
171                entity,
172                "create",
173                &row,
174                None,
175                caller_user_id,
176                schema_override,
177            )
178            .await?;
179        }
180        Ok(row)
181    }
182
183    /// Insert a parent row and its FK-children atomically in a single transaction.
184    ///
185    /// `children` pairs each `ToMany` include spec with its resolved child entity and the
186    /// list of child bodies to insert. For every child, the FK column (`spec.their_key_column`)
187    /// is set to the parent's `spec.our_key_column` value before insertion. Any error rolls
188    /// the whole transaction back, so no orphan parent or partial child set is ever committed.
189    ///
190    /// `set_local_sql` is the `SET LOCAL app.tenant_id = '...'` statement for RLS tenants; it is
191    /// run inside the transaction so it scopes to these inserts. Pass `None` for pool strategy.
192    ///
193    /// Returns `(parent_row, child_rows_by_include_name)` as raw DB rows (snake_case keys,
194    /// sensitive columns NOT stripped — the caller shapes the response).
195    #[allow(clippy::too_many_arguments)]
196    pub async fn create_graph(
197        pool: &Pool,
198        parent: &ResolvedEntity,
199        parent_body: &HashMap<String, Value>,
200        children: &[GraphChild],
201        schema_override: Option<&str>,
202        rls_tenant_id: Option<&str>,
203        set_local_sql: Option<&str>,
204        caller_user_id: Option<&str>,
205        dialect: &dyn Dialect,
206    ) -> Result<(Value, HashMap<String, Vec<Value>>), AppError> {
207        let mut tx = pool.begin().await?;
208        // RLS: scope the tenant to THIS transaction (SET LOCAL only lasts the transaction).
209        if let Some(sql) = set_local_sql {
210            sqlx::query(sql).execute(&mut *tx).await?;
211        }
212
213        let parent_row;
214        let mut child_rows: HashMap<String, Vec<Value>> = HashMap::new();
215        {
216            // A Transaction derefs to &mut Connection — the Conn executor variant handles it,
217            // so every create() below runs on this same connection (one atomic transaction).
218            let mut exec = TenantExecutor::conn(&mut tx, dialect);
219
220            parent_row = Self::create(
221                &mut exec,
222                parent,
223                parent_body,
224                schema_override,
225                rls_tenant_id,
226                caller_user_id,
227                dialect,
228            )
229            .await?;
230
231            for (spec, child_entity, bodies) in children {
232                // Value to copy from the new parent into each child's FK column.
233                let fk_value = parent_row
234                    .get(&spec.our_key_column)
235                    .cloned()
236                    .ok_or_else(|| {
237                        AppError::BadRequest(format!(
238                            "parent row is missing key column '{}' for include '{}'",
239                            spec.our_key_column, spec.name
240                        ))
241                    })?;
242                let mut rows = Vec::with_capacity(bodies.len());
243                for body in bodies {
244                    let mut child = body.clone();
245                    child.insert(spec.their_key_column.clone(), fk_value.clone());
246                    let row = Self::create(
247                        &mut exec,
248                        child_entity,
249                        &child,
250                        schema_override,
251                        rls_tenant_id,
252                        caller_user_id,
253                        dialect,
254                    )
255                    .await?;
256                    rows.push(row);
257                }
258                child_rows.insert(spec.name.clone(), rows);
259            }
260        } // exec dropped here, releasing the &mut borrow on tx
261
262        tx.commit().await?; // nothing is durable until this line
263        Ok((parent_row, child_rows))
264    }
265
266    /// Update one row by id. Returns updated row.
267    /// When caller_user_id is Some, updated_by is set to that value.
268    /// When entity has versioning enabled, a history snapshot is written atomically before the update.
269    pub async fn update<'a>(
270        executor: &mut TenantExecutor<'a>,
271        entity: &ResolvedEntity,
272        id: &Value,
273        body: &HashMap<String, Value>,
274        schema_override: Option<&str>,
275        caller_user_id: Option<&str>,
276        dialect: &dyn Dialect,
277    ) -> Result<Option<Value>, AppError> {
278        let versioning_enabled = entity.versioning.as_ref().is_some_and(|v| v.enabled);
279
280        let pre_row = if entity.audit_log || versioning_enabled {
281            let q = select_by_id(entity, schema_override, dialect);
282            Self::query_one_exec(executor, &q.sql, std::slice::from_ref(id)).await?
283        } else {
284            None
285        };
286
287        let result = if versioning_enabled {
288            // Write snapshot + update in a single transaction.
289            let snap_q = insert_history_snapshot(entity, "update", schema_override, dialect);
290            let upd_q = update(entity, id, body, schema_override, caller_user_id, dialect);
291            let keep = entity.versioning.as_ref().and_then(|v| v.keep_versions);
292            let prune_q = keep.map(|_| prune_history(entity, schema_override, dialect));
293            Self::run_versioned_update(executor, id, snap_q, upd_q, prune_q, keep).await?
294        } else {
295            let q = update(entity, id, body, schema_override, caller_user_id, dialect);
296            Self::execute_returning_one_exec(executor, &q).await?
297        };
298
299        if entity.audit_log {
300            if let Some(ref post_row) = result {
301                Self::insert_audit(
302                    executor,
303                    entity,
304                    "update",
305                    post_row,
306                    pre_row.as_ref(),
307                    caller_user_id,
308                    schema_override,
309                )
310                .await?;
311            }
312        }
313        Ok(result)
314    }
315
316    /// Delete one row by id. Returns deleted row or None.
317    /// When caller_user_id is Some, audit_by is set on the audit record.
318    /// When entity has versioning enabled, a history snapshot is written atomically before the delete.
319    pub async fn delete<'a>(
320        executor: &mut TenantExecutor<'a>,
321        entity: &ResolvedEntity,
322        id: &Value,
323        schema_override: Option<&str>,
324        caller_user_id: Option<&str>,
325        dialect: &dyn Dialect,
326    ) -> Result<Option<Value>, AppError> {
327        let versioning_enabled = entity.versioning.as_ref().is_some_and(|v| v.enabled);
328
329        let result = if versioning_enabled {
330            let snap_q = insert_history_snapshot(entity, "delete", schema_override, dialect);
331            let del_q = delete(entity, schema_override, dialect);
332            Self::run_versioned_delete(executor, id, snap_q, del_q).await?
333        } else {
334            let q = delete(entity, schema_override, dialect);
335            Self::execute_returning_one_with_params_exec(executor, &q.sql, std::slice::from_ref(id))
336                .await?
337        };
338
339        if entity.audit_log {
340            if let Some(ref deleted_row) = result {
341                Self::insert_audit(
342                    executor,
343                    entity,
344                    "delete",
345                    deleted_row,
346                    None,
347                    caller_user_id,
348                    schema_override,
349                )
350                .await?;
351            }
352        }
353        Ok(result)
354    }
355
356    /// Archive one row by id: stamps archive_field with NOW() if it is currently NULL.
357    /// Returns the updated row, or None if the record was not found or already archived.
358    pub async fn archive<'a>(
359        executor: &mut TenantExecutor<'a>,
360        entity: &ResolvedEntity,
361        archive_field: &str,
362        id: &Value,
363        schema_override: Option<&str>,
364        dialect: &dyn Dialect,
365    ) -> Result<Option<Value>, AppError> {
366        let q = archive(entity, archive_field, schema_override, dialect);
367        Self::execute_returning_one_with_params_exec(executor, &q.sql, std::slice::from_ref(id))
368            .await
369    }
370
371    /// Unarchive one row by id: clears archive_field (sets to NULL) if it is currently NOT NULL.
372    /// Returns the updated row, or None if the record was not found or not archived.
373    pub async fn unarchive<'a>(
374        executor: &mut TenantExecutor<'a>,
375        entity: &ResolvedEntity,
376        archive_field: &str,
377        id: &Value,
378        schema_override: Option<&str>,
379        dialect: &dyn Dialect,
380    ) -> Result<Option<Value>, AppError> {
381        let q = unarchive(entity, archive_field, schema_override, dialect);
382        Self::execute_returning_one_with_params_exec(executor, &q.sql, std::slice::from_ref(id))
383            .await
384    }
385
386    /// Bulk create in a transaction (when using pool) or on the same connection (when using conn). Returns vec of created rows.
387    /// When rls_tenant_id is Some (RLS strategy), tenant_id column is set automatically on each row.
388    /// When caller_user_id is Some, created_by is set on each row.
389    pub async fn bulk_create<'a>(
390        executor: &mut TenantExecutor<'a>,
391        entity: &ResolvedEntity,
392        items: &[HashMap<String, Value>],
393        schema_override: Option<&str>,
394        rls_tenant_id: Option<&str>,
395        caller_user_id: Option<&str>,
396        dialect: &dyn Dialect,
397    ) -> Result<Vec<Value>, AppError> {
398        const BULK_LIMIT: usize = 100;
399        if items.len() > BULK_LIMIT {
400            return Err(AppError::BadRequest(format!(
401                "bulk create limited to {} items",
402                BULK_LIMIT
403            )));
404        }
405        let mut out = Vec::with_capacity(items.len());
406        match executor.executor {
407            TenantExecutorInner::Pool(pool) => {
408                let mut tx = pool.begin().await?;
409                for body in items {
410                    let include_pk = body.contains_key(&entity.pk_columns[0]);
411                    let q = insert(
412                        entity,
413                        body,
414                        include_pk,
415                        schema_override,
416                        rls_tenant_id,
417                        caller_user_id,
418                        dialect,
419                    );
420                    let row = Self::execute_returning_one_tx(&mut tx, &q)
421                        .await?
422                        .unwrap_or(Value::Null);
423                    out.push(row);
424                }
425                tx.commit().await?;
426            }
427            TenantExecutorInner::Conn(ref mut conn) => {
428                for body in items {
429                    let include_pk = body.contains_key(&entity.pk_columns[0]);
430                    let q = insert(
431                        entity,
432                        body,
433                        include_pk,
434                        schema_override,
435                        rls_tenant_id,
436                        caller_user_id,
437                        dialect,
438                    );
439                    let row = Self::execute_returning_one_conn(conn, &q)
440                        .await?
441                        .unwrap_or(Value::Null);
442                    out.push(row);
443                }
444            }
445        }
446        Ok(out)
447    }
448
449    /// Like `bulk_create` but uses savepoints to isolate per-row DB errors.
450    /// Returns `(successful_rows, row_errors)`. If any errors occur the transaction is
451    /// rolled back and successful_rows will be empty — call site decides how to surface errors.
452    pub async fn bulk_create_collecting<'a>(
453        executor: &mut TenantExecutor<'a>,
454        entity: &ResolvedEntity,
455        items: &[HashMap<String, Value>],
456        schema_override: Option<&str>,
457        rls_tenant_id: Option<&str>,
458        caller_user_id: Option<&str>,
459        dialect: &dyn Dialect,
460    ) -> Result<(Vec<Value>, Vec<(usize, AppError)>), AppError> {
461        const BULK_LIMIT: usize = 100;
462        if items.len() > BULK_LIMIT {
463            return Err(AppError::BadRequest(format!(
464                "bulk create limited to {} items",
465                BULK_LIMIT
466            )));
467        }
468        let mut out = Vec::with_capacity(items.len());
469        let mut row_errors: Vec<(usize, AppError)> = Vec::new();
470        match executor.executor {
471            TenantExecutorInner::Pool(pool) => {
472                let mut tx = pool.begin().await?;
473                for (idx, body) in items.iter().enumerate() {
474                    let sp = format!("sp_{}", idx);
475                    sqlx::query(&format!("SAVEPOINT {}", sp))
476                        .execute(&mut *tx)
477                        .await?;
478                    let include_pk = body.contains_key(&entity.pk_columns[0]);
479                    let q = insert(
480                        entity,
481                        body,
482                        include_pk,
483                        schema_override,
484                        rls_tenant_id,
485                        caller_user_id,
486                        dialect,
487                    );
488                    match Self::execute_returning_one_tx(&mut tx, &q).await {
489                        Ok(row) => {
490                            sqlx::query(&format!("RELEASE SAVEPOINT {}", sp))
491                                .execute(&mut *tx)
492                                .await?;
493                            out.push(row.unwrap_or(Value::Null));
494                        }
495                        Err(e) => {
496                            sqlx::query(&format!("ROLLBACK TO SAVEPOINT {}", sp))
497                                .execute(&mut *tx)
498                                .await?;
499                            row_errors.push((idx, e));
500                        }
501                    }
502                }
503                if row_errors.is_empty() {
504                    tx.commit().await?;
505                } else {
506                    tx.rollback().await?;
507                    out.clear();
508                }
509            }
510            TenantExecutorInner::Conn(ref mut conn) => {
511                for (idx, body) in items.iter().enumerate() {
512                    let sp = format!("sp_{}", idx);
513                    sqlx::query(&format!("SAVEPOINT {}", sp))
514                        .execute(&mut **conn)
515                        .await?;
516                    let include_pk = body.contains_key(&entity.pk_columns[0]);
517                    let q = insert(
518                        entity,
519                        body,
520                        include_pk,
521                        schema_override,
522                        rls_tenant_id,
523                        caller_user_id,
524                        dialect,
525                    );
526                    match Self::execute_returning_one_conn(conn, &q).await {
527                        Ok(row) => {
528                            sqlx::query(&format!("RELEASE SAVEPOINT {}", sp))
529                                .execute(&mut **conn)
530                                .await?;
531                            out.push(row.unwrap_or(Value::Null));
532                        }
533                        Err(e) => {
534                            sqlx::query(&format!("ROLLBACK TO SAVEPOINT {}", sp))
535                                .execute(&mut **conn)
536                                .await?;
537                            row_errors.push((idx, e));
538                        }
539                    }
540                }
541                if !row_errors.is_empty() {
542                    out.clear();
543                }
544            }
545        }
546        Ok((out, row_errors))
547    }
548
549    /// Bulk update in a transaction (when using pool) or on the same connection (when using conn). Each item must have id. Returns vec of updated rows.
550    /// When caller_user_id is Some, updated_by is set on each row.
551    pub async fn bulk_update<'a>(
552        executor: &mut TenantExecutor<'a>,
553        entity: &ResolvedEntity,
554        items: &[HashMap<String, Value>],
555        schema_override: Option<&str>,
556        caller_user_id: Option<&str>,
557        dialect: &dyn Dialect,
558    ) -> Result<Vec<Value>, AppError> {
559        const BULK_LIMIT: usize = 100;
560        if items.len() > BULK_LIMIT {
561            return Err(AppError::BadRequest(format!(
562                "bulk update limited to {} items",
563                BULK_LIMIT
564            )));
565        }
566        let pk = &entity.pk_columns[0];
567        let mut out = Vec::with_capacity(items.len());
568        match executor.executor {
569            TenantExecutorInner::Pool(pool) => {
570                let mut tx = pool.begin().await?;
571                for body in items {
572                    let id = body.get(pk).ok_or_else(|| {
573                        AppError::Validation(format!("each item must have '{}'", pk))
574                    })?;
575                    let mut body_without_pk = body.clone();
576                    body_without_pk.remove(pk);
577                    let q = update(
578                        entity,
579                        id,
580                        &body_without_pk,
581                        schema_override,
582                        caller_user_id,
583                        dialect,
584                    );
585                    if let Some(row) = Self::execute_returning_one_tx(&mut tx, &q).await? {
586                        out.push(row);
587                    }
588                }
589                tx.commit().await?;
590            }
591            TenantExecutorInner::Conn(ref mut conn) => {
592                for body in items {
593                    let id = body.get(pk).ok_or_else(|| {
594                        AppError::Validation(format!("each item must have '{}'", pk))
595                    })?;
596                    let mut body_without_pk = body.clone();
597                    body_without_pk.remove(pk);
598                    let q = update(
599                        entity,
600                        id,
601                        &body_without_pk,
602                        schema_override,
603                        caller_user_id,
604                        dialect,
605                    );
606                    if let Some(row) = Self::execute_returning_one_conn(conn, &q).await? {
607                        out.push(row);
608                    }
609                }
610            }
611        }
612        Ok(out)
613    }
614
615    /// Like `bulk_update` but uses savepoints to isolate per-row DB errors.
616    /// Missing pk on an item is recorded as a row error rather than aborting early.
617    /// Returns `(successful_rows, row_errors)`. If any errors occur the transaction is
618    /// rolled back and successful_rows will be empty.
619    pub async fn bulk_update_collecting<'a>(
620        executor: &mut TenantExecutor<'a>,
621        entity: &ResolvedEntity,
622        items: &[HashMap<String, Value>],
623        schema_override: Option<&str>,
624        caller_user_id: Option<&str>,
625        dialect: &dyn Dialect,
626    ) -> Result<(Vec<Value>, Vec<(usize, AppError)>), AppError> {
627        const BULK_LIMIT: usize = 100;
628        if items.len() > BULK_LIMIT {
629            return Err(AppError::BadRequest(format!(
630                "bulk update limited to {} items",
631                BULK_LIMIT
632            )));
633        }
634        let pk = entity.pk_columns[0].clone();
635        let mut out = Vec::with_capacity(items.len());
636        let mut row_errors: Vec<(usize, AppError)> = Vec::new();
637        match executor.executor {
638            TenantExecutorInner::Pool(pool) => {
639                let mut tx = pool.begin().await?;
640                for (idx, body) in items.iter().enumerate() {
641                    let id = match body.get(&pk) {
642                        Some(id) => id.clone(),
643                        None => {
644                            row_errors.push((
645                                idx,
646                                AppError::Validation(format!("each item must have '{}'", pk)),
647                            ));
648                            continue;
649                        }
650                    };
651                    let sp = format!("sp_{}", idx);
652                    sqlx::query(&format!("SAVEPOINT {}", sp))
653                        .execute(&mut *tx)
654                        .await?;
655                    let mut body_without_pk = body.clone();
656                    body_without_pk.remove(&pk);
657                    let q = update(
658                        entity,
659                        &id,
660                        &body_without_pk,
661                        schema_override,
662                        caller_user_id,
663                        dialect,
664                    );
665                    match Self::execute_returning_one_tx(&mut tx, &q).await {
666                        Ok(Some(row)) => {
667                            sqlx::query(&format!("RELEASE SAVEPOINT {}", sp))
668                                .execute(&mut *tx)
669                                .await?;
670                            out.push(row);
671                        }
672                        Ok(None) => {
673                            sqlx::query(&format!("RELEASE SAVEPOINT {}", sp))
674                                .execute(&mut *tx)
675                                .await?;
676                        }
677                        Err(e) => {
678                            sqlx::query(&format!("ROLLBACK TO SAVEPOINT {}", sp))
679                                .execute(&mut *tx)
680                                .await?;
681                            row_errors.push((idx, e));
682                        }
683                    }
684                }
685                if row_errors.is_empty() {
686                    tx.commit().await?;
687                } else {
688                    tx.rollback().await?;
689                    out.clear();
690                }
691            }
692            TenantExecutorInner::Conn(ref mut conn) => {
693                for (idx, body) in items.iter().enumerate() {
694                    let id = match body.get(&pk) {
695                        Some(id) => id.clone(),
696                        None => {
697                            row_errors.push((
698                                idx,
699                                AppError::Validation(format!("each item must have '{}'", pk)),
700                            ));
701                            continue;
702                        }
703                    };
704                    let sp = format!("sp_{}", idx);
705                    sqlx::query(&format!("SAVEPOINT {}", sp))
706                        .execute(&mut **conn)
707                        .await?;
708                    let mut body_without_pk = body.clone();
709                    body_without_pk.remove(&pk);
710                    let q = update(
711                        entity,
712                        &id,
713                        &body_without_pk,
714                        schema_override,
715                        caller_user_id,
716                        dialect,
717                    );
718                    match Self::execute_returning_one_conn(conn, &q).await {
719                        Ok(Some(row)) => {
720                            sqlx::query(&format!("RELEASE SAVEPOINT {}", sp))
721                                .execute(&mut **conn)
722                                .await?;
723                            out.push(row);
724                        }
725                        Ok(None) => {
726                            sqlx::query(&format!("RELEASE SAVEPOINT {}", sp))
727                                .execute(&mut **conn)
728                                .await?;
729                        }
730                        Err(e) => {
731                            sqlx::query(&format!("ROLLBACK TO SAVEPOINT {}", sp))
732                                .execute(&mut **conn)
733                                .await?;
734                            row_errors.push((idx, e));
735                        }
736                    }
737                }
738                if !row_errors.is_empty() {
739                    out.clear();
740                }
741            }
742        }
743        Ok((out, row_errors))
744    }
745
746    /// Execute a history SELECT that returns multiple rows (used by list_history handler).
747    /// Binds: params[0] = pk value.
748    pub async fn query_history_many<'a>(
749        executor: &mut TenantExecutor<'a>,
750        sql: &str,
751        params: &[Value],
752    ) -> Result<Vec<Value>, AppError> {
753        Self::query_many_exec(executor, sql, params).await
754    }
755
756    /// Execute a history SELECT that returns one row (used by read_history_version handler).
757    /// Binds: $1 = pk value, $2 = version (i64).
758    pub async fn query_history_one<'a>(
759        executor: &mut TenantExecutor<'a>,
760        sql: &str,
761        id: &Value,
762        version: i64,
763    ) -> Result<Option<Value>, AppError> {
764        tracing::debug!(sql = %sql, "history query");
765        let mut query = sqlx::query(sql);
766        query = query.bind(Self::to_sqlx_param(id));
767        query = query.bind(version);
768        let row = match executor.executor {
769            TenantExecutorInner::Pool(pool) => query.fetch_optional(pool).await?,
770            TenantExecutorInner::Conn(ref mut conn) => query.fetch_optional(&mut **conn).await?,
771        };
772        Ok(row.map(|r| row_to_json(&r)))
773    }
774
775    async fn query_one_exec<'a>(
776        executor: &mut TenantExecutor<'a>,
777        sql: &str,
778        params: &[Value],
779    ) -> Result<Option<Value>, AppError> {
780        tracing::debug!(sql = %sql, params = ?params, "query");
781        let bind = Self::to_sqlx_param(&params[0]);
782        let row = match executor.executor {
783            TenantExecutorInner::Pool(pool) => {
784                sqlx::query(sql).bind(bind).fetch_optional(pool).await?
785            }
786            TenantExecutorInner::Conn(ref mut conn) => {
787                sqlx::query(sql)
788                    .bind(bind)
789                    .fetch_optional(&mut **conn)
790                    .await?
791            }
792        };
793        Ok(row.map(|r| row_to_json(&r)))
794    }
795
796    async fn query_many_exec<'a>(
797        executor: &mut TenantExecutor<'a>,
798        sql: &str,
799        params: &[Value],
800    ) -> Result<Vec<Value>, AppError> {
801        tracing::debug!(sql = %sql, params = ?params, "query");
802        let mut query = sqlx::query(sql);
803        for p in params {
804            query = query.bind(Self::to_sqlx_param(p));
805        }
806        let rows = match executor.executor {
807            TenantExecutorInner::Pool(pool) => query.fetch_all(pool).await?,
808            TenantExecutorInner::Conn(ref mut conn) => query.fetch_all(&mut **conn).await?,
809        };
810        Ok(rows.iter().map(row_to_json).collect())
811    }
812
813    async fn execute_returning_one_exec<'a>(
814        executor: &mut TenantExecutor<'a>,
815        q: &QueryBuf,
816    ) -> Result<Option<Value>, AppError> {
817        tracing::debug!(sql = %q.sql, params = ?q.params, "query");
818        let mut query = sqlx::query(&q.sql);
819        for p in &q.params {
820            query = query.bind(Self::to_sqlx_param(p));
821        }
822        let row = match executor.executor {
823            TenantExecutorInner::Pool(pool) => query.fetch_optional(pool).await?,
824            TenantExecutorInner::Conn(ref mut conn) => query.fetch_optional(&mut **conn).await?,
825        };
826        Ok(row.map(|r| row_to_json(&r)))
827    }
828
829    async fn execute_returning_one_with_params_exec<'a>(
830        executor: &mut TenantExecutor<'a>,
831        sql: &str,
832        params: &[Value],
833    ) -> Result<Option<Value>, AppError> {
834        tracing::debug!(sql = %sql, params = ?params, "query");
835        let mut query = sqlx::query(sql);
836        for p in params {
837            query = query.bind(Self::to_sqlx_param(p));
838        }
839        let row = match executor.executor {
840            TenantExecutorInner::Pool(pool) => query.fetch_optional(pool).await?,
841            TenantExecutorInner::Conn(ref mut conn) => query.fetch_optional(&mut **conn).await?,
842        };
843        Ok(row.map(|r| row_to_json(&r)))
844    }
845
846    async fn execute_returning_one_conn(
847        conn: &mut Connection,
848        q: &QueryBuf,
849    ) -> Result<Option<Value>, AppError> {
850        tracing::debug!(sql = %q.sql, params = ?q.params, "query (conn)");
851        let mut query = sqlx::query(&q.sql);
852        for p in &q.params {
853            query = query.bind(Self::to_sqlx_param(p));
854        }
855        let row = query.fetch_optional(conn).await?;
856        Ok(row.map(|r| row_to_json(&r)))
857    }
858
859    async fn execute_returning_one_tx(
860        tx: &mut Connection,
861        q: &QueryBuf,
862    ) -> Result<Option<Value>, AppError> {
863        tracing::debug!(sql = %q.sql, params = ?q.params, "query (tx)");
864        let mut query = sqlx::query(&q.sql);
865        for p in &q.params {
866            query = query.bind(Self::to_sqlx_param(p));
867        }
868        let row = query.fetch_optional(&mut *tx).await?;
869        Ok(row.map(|r| row_to_json(&r)))
870    }
871
872    fn to_sqlx_param(v: &Value) -> BindValue {
873        BindValue::from_json(v).unwrap_or(BindValue::Null)
874    }
875
876    /// Snapshot + UPDATE in one transaction (versioning path for update).
877    async fn run_versioned_update<'a>(
878        executor: &mut TenantExecutor<'a>,
879        id: &Value,
880        snap_q: QueryBuf,
881        upd_q: QueryBuf,
882        prune_q: Option<QueryBuf>,
883        keep_versions: Option<i64>,
884    ) -> Result<Option<Value>, AppError> {
885        match executor.executor {
886            TenantExecutorInner::Pool(pool) => {
887                let mut tx = pool.begin().await?;
888                // Snapshot (INSERT INTO _history SELECT ...)
889                let mut snap = sqlx::query(&snap_q.sql);
890                snap = snap.bind(Self::to_sqlx_param(&snap_q.params[0])); // operation
891                snap = snap.bind(Self::to_sqlx_param(id)); // pk
892                snap.execute(&mut *tx).await?;
893                // Update
894                let mut upd = sqlx::query(&upd_q.sql);
895                for p in &upd_q.params {
896                    upd = upd.bind(Self::to_sqlx_param(p));
897                }
898                let row = upd.fetch_optional(&mut *tx).await?.map(|r| row_to_json(&r));
899                // Prune
900                if let (Some(pq), Some(kv)) = (prune_q, keep_versions) {
901                    let mut pr = sqlx::query(&pq.sql);
902                    pr = pr.bind(Self::to_sqlx_param(id));
903                    pr = pr.bind(kv);
904                    pr.execute(&mut *tx).await?;
905                }
906                tx.commit().await?;
907                Ok(row)
908            }
909            TenantExecutorInner::Conn(ref mut conn) => {
910                // On an RLS connection we can't open a nested transaction; use SAVEPOINT.
911                sqlx::query("SAVEPOINT sp_versioned_update")
912                    .execute(&mut **conn)
913                    .await?;
914                let snap_res = async {
915                    let mut snap = sqlx::query(&snap_q.sql);
916                    snap = snap.bind(Self::to_sqlx_param(&snap_q.params[0]));
917                    snap = snap.bind(Self::to_sqlx_param(id));
918                    snap.execute(&mut **conn).await?;
919                    let mut upd = sqlx::query(&upd_q.sql);
920                    for p in &upd_q.params {
921                        upd = upd.bind(Self::to_sqlx_param(p));
922                    }
923                    let row = upd
924                        .fetch_optional(&mut **conn)
925                        .await?
926                        .map(|r| row_to_json(&r));
927                    if let (Some(pq), Some(kv)) = (prune_q, keep_versions) {
928                        let mut pr = sqlx::query(&pq.sql);
929                        pr = pr.bind(Self::to_sqlx_param(id));
930                        pr = pr.bind(kv);
931                        pr.execute(&mut **conn).await?;
932                    }
933                    Ok::<_, sqlx::Error>(row)
934                }
935                .await;
936                match snap_res {
937                    Ok(row) => {
938                        sqlx::query("RELEASE SAVEPOINT sp_versioned_update")
939                            .execute(&mut **conn)
940                            .await?;
941                        Ok(row)
942                    }
943                    Err(e) => {
944                        sqlx::query("ROLLBACK TO SAVEPOINT sp_versioned_update")
945                            .execute(&mut **conn)
946                            .await?;
947                        Err(AppError::Db(e))
948                    }
949                }
950            }
951        }
952    }
953
954    /// Snapshot + DELETE in one transaction (versioning path for delete).
955    async fn run_versioned_delete<'a>(
956        executor: &mut TenantExecutor<'a>,
957        id: &Value,
958        snap_q: QueryBuf,
959        del_q: QueryBuf,
960    ) -> Result<Option<Value>, AppError> {
961        match executor.executor {
962            TenantExecutorInner::Pool(pool) => {
963                let mut tx = pool.begin().await?;
964                let mut snap = sqlx::query(&snap_q.sql);
965                snap = snap.bind(Self::to_sqlx_param(&snap_q.params[0])); // operation
966                snap = snap.bind(Self::to_sqlx_param(id)); // pk
967                snap.execute(&mut *tx).await?;
968                let mut del = sqlx::query(&del_q.sql);
969                del = del.bind(Self::to_sqlx_param(id));
970                let row = del.fetch_optional(&mut *tx).await?.map(|r| row_to_json(&r));
971                tx.commit().await?;
972                Ok(row)
973            }
974            TenantExecutorInner::Conn(ref mut conn) => {
975                sqlx::query("SAVEPOINT sp_versioned_delete")
976                    .execute(&mut **conn)
977                    .await?;
978                let snap_res = async {
979                    let mut snap = sqlx::query(&snap_q.sql);
980                    snap = snap.bind(Self::to_sqlx_param(&snap_q.params[0]));
981                    snap = snap.bind(Self::to_sqlx_param(id));
982                    snap.execute(&mut **conn).await?;
983                    let mut del = sqlx::query(&del_q.sql);
984                    del = del.bind(Self::to_sqlx_param(id));
985                    let row = del
986                        .fetch_optional(&mut **conn)
987                        .await?
988                        .map(|r| row_to_json(&r));
989                    Ok::<_, sqlx::Error>(row)
990                }
991                .await;
992                match snap_res {
993                    Ok(row) => {
994                        sqlx::query("RELEASE SAVEPOINT sp_versioned_delete")
995                            .execute(&mut **conn)
996                            .await?;
997                        Ok(row)
998                    }
999                    Err(e) => {
1000                        sqlx::query("ROLLBACK TO SAVEPOINT sp_versioned_delete")
1001                            .execute(&mut **conn)
1002                            .await?;
1003                        Err(AppError::Db(e))
1004                    }
1005                }
1006            }
1007        }
1008    }
1009
1010    async fn insert_audit<'a>(
1011        executor: &mut TenantExecutor<'a>,
1012        entity: &ResolvedEntity,
1013        action: &str,
1014        row: &Value,
1015        pre_row: Option<&Value>,
1016        audit_by: Option<&str>,
1017        schema_override: Option<&str>,
1018    ) -> Result<(), AppError> {
1019        let schema = schema_override.unwrap_or(&entity.schema_name);
1020        let audit_table = format!(
1021            "\"{}\".\"{}\"",
1022            schema.replace('"', "\"\""),
1023            format!("{}_audit", entity.table_name).replace('"', "\"\"")
1024        );
1025
1026        let changed = if action == "update" {
1027            pre_row.map(|pre| compute_changed_fields(pre, row, entity))
1028        } else {
1029            None
1030        };
1031
1032        let mut col_names: Vec<String> = vec![
1033            "\"audit_action\"".to_string(),
1034            "\"audit_by\"".to_string(),
1035            "\"changed_fields\"".to_string(),
1036        ];
1037        let mut placeholders: Vec<String> = Vec::new();
1038        let mut params: Vec<Value> = Vec::new();
1039
1040        params.push(Value::String(action.to_string()));
1041        placeholders.push(format!("${}", params.len()));
1042
1043        params.push(
1044            audit_by
1045                .map(|s| Value::String(s.to_string()))
1046                .unwrap_or(Value::Null),
1047        );
1048        placeholders.push(format!("${}", params.len()));
1049
1050        params.push(changed.unwrap_or(Value::Null));
1051        placeholders.push(format!("${}::jsonb", params.len()));
1052
1053        let row_obj = row.as_object();
1054        for col in &entity.columns {
1055            let raw = row_obj
1056                .and_then(|o| o.get(&col.name))
1057                .cloned()
1058                .unwrap_or(Value::Null);
1059            let val = coerce_json_value_for_pg_array(raw, col.pg_type.as_deref());
1060            let param_num = params.len() + 1;
1061            let ph = col
1062                .pg_type
1063                .as_deref()
1064                .map(|t| format!("${}::{}", param_num, t))
1065                .unwrap_or_else(|| format!("${}", param_num));
1066            col_names.push(format!("\"{}\"", col.name));
1067            placeholders.push(ph);
1068            params.push(val);
1069        }
1070
1071        let sql = format!(
1072            "INSERT INTO {} ({}) VALUES ({})",
1073            audit_table,
1074            col_names.join(", "),
1075            placeholders.join(", ")
1076        );
1077        tracing::debug!(sql = %sql, "audit insert");
1078
1079        let mut query = sqlx::query(&sql);
1080        for p in &params {
1081            query = query.bind(Self::to_sqlx_param(p));
1082        }
1083        match executor.executor {
1084            TenantExecutorInner::Pool(pool) => {
1085                query.execute(pool).await?;
1086            }
1087            TenantExecutorInner::Conn(ref mut conn) => {
1088                query.execute(&mut **conn).await?;
1089            }
1090        }
1091        Ok(())
1092    }
1093}
1094
1095fn compute_changed_fields(pre: &Value, post: &Value, entity: &ResolvedEntity) -> Value {
1096    let pre_obj = match pre.as_object() {
1097        Some(o) => o,
1098        None => return Value::Null,
1099    };
1100    let post_obj = match post.as_object() {
1101        Some(o) => o,
1102        None => return Value::Null,
1103    };
1104    let mut changes = serde_json::Map::new();
1105    for col in &entity.columns {
1106        let pre_val = pre_obj.get(&col.name).unwrap_or(&Value::Null);
1107        let post_val = post_obj.get(&col.name).unwrap_or(&Value::Null);
1108        if pre_val != post_val {
1109            let mut diff = serde_json::Map::new();
1110            diff.insert("old".to_string(), pre_val.clone());
1111            diff.insert("new".to_string(), post_val.clone());
1112            changes.insert(col.name.clone(), Value::Object(diff));
1113        }
1114    }
1115    Value::Object(changes)
1116}
1117
1118fn row_to_json(row: &DbRow) -> Value {
1119    use sqlx::Column;
1120    use sqlx::Row;
1121    let mut map = serde_json::Map::new();
1122    for col in row.columns() {
1123        let name = col.name();
1124        let v = cell_to_value(row, name);
1125        map.insert(name.to_string(), v);
1126    }
1127    Value::Object(map)
1128}
1129
1130fn cell_to_value(row: &DbRow, name: &str) -> Value {
1131    use sqlx::Row;
1132    if let Ok(Some(n)) = row.try_get::<Option<i16>, _>(name) {
1133        return Value::Number(n.into());
1134    }
1135    if let Ok(Some(n)) = row.try_get::<Option<i32>, _>(name) {
1136        return Value::Number(n.into());
1137    }
1138    if let Ok(Some(n)) = row.try_get::<Option<i64>, _>(name) {
1139        return Value::Number(n.into());
1140    }
1141    if let Ok(Some(n)) = row.try_get::<Option<f32>, _>(name) {
1142        if let Some(n) = serde_json::Number::from_f64(n as f64) {
1143            return Value::Number(n);
1144        }
1145    }
1146    if let Ok(Some(n)) = row.try_get::<Option<f64>, _>(name) {
1147        if let Some(n) = serde_json::Number::from_f64(n) {
1148            return Value::Number(n);
1149        }
1150    }
1151    if let Ok(Some(b)) = row.try_get::<Option<bool>, _>(name) {
1152        return Value::Bool(b);
1153    }
1154    #[cfg(feature = "postgres")]
1155    if let Ok(Some(vec)) = row.try_get::<Option<Vec<String>>, _>(name) {
1156        return Value::Array(vec.into_iter().map(Value::String).collect());
1157    }
1158    #[cfg(feature = "postgres")]
1159    if let Ok(Some(vec)) = row.try_get::<Option<Vec<uuid::Uuid>>, _>(name) {
1160        return Value::Array(
1161            vec.into_iter()
1162                .map(|u| Value::String(u.to_string()))
1163                .collect(),
1164        );
1165    }
1166    #[cfg(feature = "postgres")]
1167    if let Ok(Some(vec)) = row.try_get::<Option<Vec<i64>>, _>(name) {
1168        return Value::Array(vec.into_iter().map(|n| Value::Number(n.into())).collect());
1169    }
1170    if let Ok(Some(u)) = row.try_get::<Option<uuid::Uuid>, _>(name) {
1171        return Value::String(u.to_string());
1172    }
1173    if let Ok(Some(d)) = row.try_get::<Option<chrono::DateTime<chrono::Utc>>, _>(name) {
1174        return Value::String(d.to_rfc3339());
1175    }
1176    if let Ok(Some(d)) = row.try_get::<Option<chrono::NaiveDateTime>, _>(name) {
1177        return Value::String(d.format("%Y-%m-%dT%H:%M:%S%.f").to_string());
1178    }
1179    if let Ok(Some(d)) = row.try_get::<Option<chrono::NaiveDate>, _>(name) {
1180        return Value::String(d.format("%Y-%m-%d").to_string());
1181    }
1182    if let Ok(Some(s)) = row.try_get::<Option<String>, _>(name) {
1183        // Numeric columns are selected as ::text; parse so we return a JSON number not string
1184        if let Ok(n) = s.trim().parse::<f64>() {
1185            if let Some(num) = serde_json::Number::from_f64(n) {
1186                return Value::Number(num);
1187            }
1188        }
1189        return Value::String(s);
1190    }
1191    if let Ok(Some(j)) = row.try_get::<Option<serde_json::Value>, _>(name) {
1192        return j;
1193    }
1194    Value::Null
1195}
architect_sdk/service/crud.rs

architect_sdk/service/
crud.rs
