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

architect_sdk/service/
crud.rs
