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    async fn query_many_exec<'a>(
995        executor: &mut TenantExecutor<'a>,
996        sql: &str,
997        params: &[Value],
998    ) -> Result<Vec<Value>, AppError> {
999        tracing::debug!(sql = %sql, params = ?params, "query");
1000        let mut query = sqlx::query(sql);
1001        for p in params {
1002            query = query.bind(Self::to_sqlx_param(p));
1003        }
1004        let rows = match executor.executor {
1005            TenantExecutorInner::Pool(pool) => query.fetch_all(pool).await?,
1006            TenantExecutorInner::Conn(ref mut conn) => query.fetch_all(&mut **conn).await?,
1007        };
1008        Ok(rows.iter().map(row_to_json).collect())
1009    }
1010
1011    async fn execute_returning_one_exec<'a>(
1012        executor: &mut TenantExecutor<'a>,
1013        q: &QueryBuf,
1014    ) -> Result<Option<Value>, AppError> {
1015        tracing::debug!(sql = %q.sql, params = ?q.params, "query");
1016        let mut query = sqlx::query(&q.sql);
1017        for p in &q.params {
1018            query = query.bind(Self::to_sqlx_param(p));
1019        }
1020        let row = match executor.executor {
1021            TenantExecutorInner::Pool(pool) => query.fetch_optional(pool).await?,
1022            TenantExecutorInner::Conn(ref mut conn) => query.fetch_optional(&mut **conn).await?,
1023        };
1024        Ok(row.map(|r| row_to_json(&r)))
1025    }
1026
1027    async fn execute_returning_one_with_params_exec<'a>(
1028        executor: &mut TenantExecutor<'a>,
1029        sql: &str,
1030        params: &[Value],
1031    ) -> Result<Option<Value>, AppError> {
1032        tracing::debug!(sql = %sql, params = ?params, "query");
1033        let mut query = sqlx::query(sql);
1034        for p in params {
1035            query = query.bind(Self::to_sqlx_param(p));
1036        }
1037        let row = match executor.executor {
1038            TenantExecutorInner::Pool(pool) => query.fetch_optional(pool).await?,
1039            TenantExecutorInner::Conn(ref mut conn) => query.fetch_optional(&mut **conn).await?,
1040        };
1041        Ok(row.map(|r| row_to_json(&r)))
1042    }
1043
1044    fn to_sqlx_param(v: &Value) -> BindValue {
1045        BindValue::from_json(v).unwrap_or(BindValue::Null)
1046    }
1047
1048    /// Snapshot + UPDATE in one transaction (versioning path for update).
1049    async fn run_versioned_update<'a>(
1050        executor: &mut TenantExecutor<'a>,
1051        id: &Value,
1052        snap_q: QueryBuf,
1053        upd_q: QueryBuf,
1054        prune_q: Option<QueryBuf>,
1055        keep_versions: Option<i64>,
1056    ) -> Result<Option<Value>, AppError> {
1057        match executor.executor {
1058            TenantExecutorInner::Pool(pool) => {
1059                let mut tx = pool.begin().await?;
1060                // Snapshot (INSERT INTO _history SELECT ...)
1061                let mut snap = sqlx::query(&snap_q.sql);
1062                snap = snap.bind(Self::to_sqlx_param(&snap_q.params[0])); // operation
1063                snap = snap.bind(Self::to_sqlx_param(id)); // pk
1064                snap.execute(&mut *tx).await?;
1065                // Update
1066                let mut upd = sqlx::query(&upd_q.sql);
1067                for p in &upd_q.params {
1068                    upd = upd.bind(Self::to_sqlx_param(p));
1069                }
1070                let row = upd.fetch_optional(&mut *tx).await?.map(|r| row_to_json(&r));
1071                // Prune
1072                if let (Some(pq), Some(kv)) = (prune_q, keep_versions) {
1073                    let mut pr = sqlx::query(&pq.sql);
1074                    pr = pr.bind(Self::to_sqlx_param(id));
1075                    pr = pr.bind(kv);
1076                    pr.execute(&mut *tx).await?;
1077                }
1078                tx.commit().await?;
1079                Ok(row)
1080            }
1081            TenantExecutorInner::Conn(ref mut conn) => {
1082                // On an RLS connection we can't open a nested transaction; use SAVEPOINT.
1083                sqlx::query("SAVEPOINT sp_versioned_update")
1084                    .execute(&mut **conn)
1085                    .await?;
1086                let snap_res = async {
1087                    let mut snap = sqlx::query(&snap_q.sql);
1088                    snap = snap.bind(Self::to_sqlx_param(&snap_q.params[0]));
1089                    snap = snap.bind(Self::to_sqlx_param(id));
1090                    snap.execute(&mut **conn).await?;
1091                    let mut upd = sqlx::query(&upd_q.sql);
1092                    for p in &upd_q.params {
1093                        upd = upd.bind(Self::to_sqlx_param(p));
1094                    }
1095                    let row = upd
1096                        .fetch_optional(&mut **conn)
1097                        .await?
1098                        .map(|r| row_to_json(&r));
1099                    if let (Some(pq), Some(kv)) = (prune_q, keep_versions) {
1100                        let mut pr = sqlx::query(&pq.sql);
1101                        pr = pr.bind(Self::to_sqlx_param(id));
1102                        pr = pr.bind(kv);
1103                        pr.execute(&mut **conn).await?;
1104                    }
1105                    Ok::<_, sqlx::Error>(row)
1106                }
1107                .await;
1108                match snap_res {
1109                    Ok(row) => {
1110                        sqlx::query("RELEASE SAVEPOINT sp_versioned_update")
1111                            .execute(&mut **conn)
1112                            .await?;
1113                        Ok(row)
1114                    }
1115                    Err(e) => {
1116                        sqlx::query("ROLLBACK TO SAVEPOINT sp_versioned_update")
1117                            .execute(&mut **conn)
1118                            .await?;
1119                        Err(AppError::Db(e))
1120                    }
1121                }
1122            }
1123        }
1124    }
1125
1126    /// Snapshot + DELETE in one transaction (versioning path for delete).
1127    async fn run_versioned_delete<'a>(
1128        executor: &mut TenantExecutor<'a>,
1129        id: &Value,
1130        snap_q: QueryBuf,
1131        del_q: QueryBuf,
1132    ) -> Result<Option<Value>, AppError> {
1133        match executor.executor {
1134            TenantExecutorInner::Pool(pool) => {
1135                let mut tx = pool.begin().await?;
1136                let mut snap = sqlx::query(&snap_q.sql);
1137                snap = snap.bind(Self::to_sqlx_param(&snap_q.params[0])); // operation
1138                snap = snap.bind(Self::to_sqlx_param(id)); // pk
1139                snap.execute(&mut *tx).await?;
1140                let mut del = sqlx::query(&del_q.sql);
1141                del = del.bind(Self::to_sqlx_param(id));
1142                let row = del.fetch_optional(&mut *tx).await?.map(|r| row_to_json(&r));
1143                tx.commit().await?;
1144                Ok(row)
1145            }
1146            TenantExecutorInner::Conn(ref mut conn) => {
1147                sqlx::query("SAVEPOINT sp_versioned_delete")
1148                    .execute(&mut **conn)
1149                    .await?;
1150                let snap_res = async {
1151                    let mut snap = sqlx::query(&snap_q.sql);
1152                    snap = snap.bind(Self::to_sqlx_param(&snap_q.params[0]));
1153                    snap = snap.bind(Self::to_sqlx_param(id));
1154                    snap.execute(&mut **conn).await?;
1155                    let mut del = sqlx::query(&del_q.sql);
1156                    del = del.bind(Self::to_sqlx_param(id));
1157                    let row = del
1158                        .fetch_optional(&mut **conn)
1159                        .await?
1160                        .map(|r| row_to_json(&r));
1161                    Ok::<_, sqlx::Error>(row)
1162                }
1163                .await;
1164                match snap_res {
1165                    Ok(row) => {
1166                        sqlx::query("RELEASE SAVEPOINT sp_versioned_delete")
1167                            .execute(&mut **conn)
1168                            .await?;
1169                        Ok(row)
1170                    }
1171                    Err(e) => {
1172                        sqlx::query("ROLLBACK TO SAVEPOINT sp_versioned_delete")
1173                            .execute(&mut **conn)
1174                            .await?;
1175                        Err(AppError::Db(e))
1176                    }
1177                }
1178            }
1179        }
1180    }
1181
1182    async fn insert_audit<'a>(
1183        executor: &mut TenantExecutor<'a>,
1184        entity: &ResolvedEntity,
1185        action: &str,
1186        row: &Value,
1187        pre_row: Option<&Value>,
1188        audit_by: Option<&str>,
1189        schema_override: Option<&str>,
1190    ) -> Result<(), AppError> {
1191        let schema = schema_override.unwrap_or(&entity.schema_name);
1192        let audit_table = format!(
1193            "\"{}\".\"{}\"",
1194            schema.replace('"', "\"\""),
1195            format!("{}_audit", entity.table_name).replace('"', "\"\"")
1196        );
1197
1198        let changed = if action == "update" {
1199            pre_row.map(|pre| compute_changed_fields(pre, row, entity))
1200        } else {
1201            None
1202        };
1203
1204        let mut col_names: Vec<String> = vec![
1205            "\"audit_action\"".to_string(),
1206            "\"audit_by\"".to_string(),
1207            "\"changed_fields\"".to_string(),
1208        ];
1209        let mut placeholders: Vec<String> = Vec::new();
1210        let mut params: Vec<Value> = Vec::new();
1211
1212        params.push(Value::String(action.to_string()));
1213        placeholders.push(format!("${}", params.len()));
1214
1215        params.push(
1216            audit_by
1217                .map(|s| Value::String(s.to_string()))
1218                .unwrap_or(Value::Null),
1219        );
1220        placeholders.push(format!("${}", params.len()));
1221
1222        params.push(changed.unwrap_or(Value::Null));
1223        placeholders.push(format!("${}::jsonb", params.len()));
1224
1225        let row_obj = row.as_object();
1226        for col in &entity.columns {
1227            let raw = row_obj
1228                .and_then(|o| o.get(&col.name))
1229                .cloned()
1230                .unwrap_or(Value::Null);
1231            let val = coerce_json_value_for_pg_array(raw, col.pg_type.as_deref());
1232            let param_num = params.len() + 1;
1233            let ph = col
1234                .pg_type
1235                .as_deref()
1236                .map(|t| format!("${}::{}", param_num, t))
1237                .unwrap_or_else(|| format!("${}", param_num));
1238            col_names.push(format!("\"{}\"", col.name));
1239            placeholders.push(ph);
1240            params.push(val);
1241        }
1242
1243        let sql = format!(
1244            "INSERT INTO {} ({}) VALUES ({})",
1245            audit_table,
1246            col_names.join(", "),
1247            placeholders.join(", ")
1248        );
1249        tracing::debug!(sql = %sql, "audit insert");
1250
1251        let mut query = sqlx::query(&sql);
1252        for p in &params {
1253            query = query.bind(Self::to_sqlx_param(p));
1254        }
1255        match executor.executor {
1256            TenantExecutorInner::Pool(pool) => {
1257                query.execute(pool).await?;
1258            }
1259            TenantExecutorInner::Conn(ref mut conn) => {
1260                query.execute(&mut **conn).await?;
1261            }
1262        }
1263        Ok(())
1264    }
1265}
1266
1267fn compute_changed_fields(pre: &Value, post: &Value, entity: &ResolvedEntity) -> Value {
1268    let pre_obj = match pre.as_object() {
1269        Some(o) => o,
1270        None => return Value::Null,
1271    };
1272    let post_obj = match post.as_object() {
1273        Some(o) => o,
1274        None => return Value::Null,
1275    };
1276    let mut changes = serde_json::Map::new();
1277    for col in &entity.columns {
1278        let pre_val = pre_obj.get(&col.name).unwrap_or(&Value::Null);
1279        let post_val = post_obj.get(&col.name).unwrap_or(&Value::Null);
1280        if pre_val != post_val {
1281            let mut diff = serde_json::Map::new();
1282            diff.insert("old".to_string(), pre_val.clone());
1283            diff.insert("new".to_string(), post_val.clone());
1284            changes.insert(col.name.clone(), Value::Object(diff));
1285        }
1286    }
1287    Value::Object(changes)
1288}
1289
1290fn row_to_json(row: &DbRow) -> Value {
1291    use sqlx::Column;
1292    use sqlx::Row;
1293    let mut map = serde_json::Map::new();
1294    for col in row.columns() {
1295        let name = col.name();
1296        let v = cell_to_value(row, name);
1297        map.insert(name.to_string(), v);
1298    }
1299    Value::Object(map)
1300}
1301
1302fn cell_to_value(row: &DbRow, name: &str) -> Value {
1303    use sqlx::Row;
1304    if let Ok(Some(n)) = row.try_get::<Option<i16>, _>(name) {
1305        return Value::Number(n.into());
1306    }
1307    if let Ok(Some(n)) = row.try_get::<Option<i32>, _>(name) {
1308        return Value::Number(n.into());
1309    }
1310    if let Ok(Some(n)) = row.try_get::<Option<i64>, _>(name) {
1311        return Value::Number(n.into());
1312    }
1313    if let Ok(Some(n)) = row.try_get::<Option<f32>, _>(name) {
1314        if let Some(n) = serde_json::Number::from_f64(n as f64) {
1315            return Value::Number(n);
1316        }
1317    }
1318    if let Ok(Some(n)) = row.try_get::<Option<f64>, _>(name) {
1319        if let Some(n) = serde_json::Number::from_f64(n) {
1320            return Value::Number(n);
1321        }
1322    }
1323    if let Ok(Some(b)) = row.try_get::<Option<bool>, _>(name) {
1324        return Value::Bool(b);
1325    }
1326    #[cfg(feature = "postgres")]
1327    if let Ok(Some(vec)) = row.try_get::<Option<Vec<String>>, _>(name) {
1328        return Value::Array(vec.into_iter().map(Value::String).collect());
1329    }
1330    #[cfg(feature = "postgres")]
1331    if let Ok(Some(vec)) = row.try_get::<Option<Vec<uuid::Uuid>>, _>(name) {
1332        return Value::Array(
1333            vec.into_iter()
1334                .map(|u| Value::String(u.to_string()))
1335                .collect(),
1336        );
1337    }
1338    #[cfg(feature = "postgres")]
1339    if let Ok(Some(vec)) = row.try_get::<Option<Vec<i64>>, _>(name) {
1340        return Value::Array(vec.into_iter().map(|n| Value::Number(n.into())).collect());
1341    }
1342    if let Ok(Some(u)) = row.try_get::<Option<uuid::Uuid>, _>(name) {
1343        return Value::String(u.to_string());
1344    }
1345    if let Ok(Some(d)) = row.try_get::<Option<chrono::DateTime<chrono::Utc>>, _>(name) {
1346        return Value::String(d.to_rfc3339());
1347    }
1348    if let Ok(Some(d)) = row.try_get::<Option<chrono::NaiveDateTime>, _>(name) {
1349        return Value::String(d.format("%Y-%m-%dT%H:%M:%S%.f").to_string());
1350    }
1351    if let Ok(Some(d)) = row.try_get::<Option<chrono::NaiveDate>, _>(name) {
1352        return Value::String(d.format("%Y-%m-%d").to_string());
1353    }
1354    if let Ok(Some(s)) = row.try_get::<Option<String>, _>(name) {
1355        // Numeric columns are selected as ::text; parse so we return a JSON number not string
1356        if let Ok(n) = s.trim().parse::<f64>() {
1357            if let Some(num) = serde_json::Number::from_f64(n) {
1358                return Value::Number(num);
1359            }
1360        }
1361        return Value::String(s);
1362    }
1363    if let Ok(Some(j)) = row.try_get::<Option<serde_json::Value>, _>(name) {
1364        return j;
1365    }
1366    Value::Null
1367}
architect_sdk/service/crud.rs

architect_sdk/service/
crud.rs
