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    pub async fn bulk_create<'a>(
455        executor: &mut TenantExecutor<'a>,
456        entity: &ResolvedEntity,
457        items: &[HashMap<String, Value>],
458        schema_override: Option<&str>,
459        rls_tenant_id: Option<&str>,
460        caller_user_id: Option<&str>,
461        dialect: &dyn Dialect,
462    ) -> Result<Vec<Value>, AppError> {
463        let bulk_limit = bulk_limit();
464        if items.len() > bulk_limit {
465            return Err(AppError::BadRequest(format!(
466                "bulk create limited to {} items",
467                bulk_limit
468            )));
469        }
470        let mut out = Vec::with_capacity(items.len());
471        match executor.executor {
472            TenantExecutorInner::Pool(pool) => {
473                let mut tx = pool.begin().await?;
474                for body in items {
475                    let include_pk = body.contains_key(&entity.pk_columns[0]);
476                    let q = insert(
477                        entity,
478                        body,
479                        include_pk,
480                        schema_override,
481                        rls_tenant_id,
482                        caller_user_id,
483                        dialect,
484                    );
485                    let row = Self::execute_returning_one_tx(&mut tx, &q)
486                        .await?
487                        .unwrap_or(Value::Null);
488                    out.push(row);
489                }
490                tx.commit().await?;
491            }
492            TenantExecutorInner::Conn(ref mut conn) => {
493                for body in items {
494                    let include_pk = body.contains_key(&entity.pk_columns[0]);
495                    let q = insert(
496                        entity,
497                        body,
498                        include_pk,
499                        schema_override,
500                        rls_tenant_id,
501                        caller_user_id,
502                        dialect,
503                    );
504                    let row = Self::execute_returning_one_conn(conn, &q)
505                        .await?
506                        .unwrap_or(Value::Null);
507                    out.push(row);
508                }
509            }
510        }
511        Ok(out)
512    }
513
514    /// Like `bulk_create` but uses savepoints to isolate per-row DB errors.
515    /// Returns `(successful_rows, row_errors)`. If any errors occur the transaction is
516    /// rolled back and successful_rows will be empty — call site decides how to surface errors.
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 include_pk = body.contains_key(&entity.pk_columns[0]);
544                    let q = insert(
545                        entity,
546                        body,
547                        include_pk,
548                        schema_override,
549                        rls_tenant_id,
550                        caller_user_id,
551                        dialect,
552                    );
553                    match Self::execute_returning_one_tx(&mut tx, &q).await {
554                        Ok(row) => {
555                            sqlx::query(&format!("RELEASE SAVEPOINT {}", sp))
556                                .execute(&mut *tx)
557                                .await?;
558                            out.push(row.unwrap_or(Value::Null));
559                        }
560                        Err(e) => {
561                            sqlx::query(&format!("ROLLBACK TO SAVEPOINT {}", sp))
562                                .execute(&mut *tx)
563                                .await?;
564                            row_errors.push((idx, e));
565                        }
566                    }
567                }
568                if row_errors.is_empty() {
569                    tx.commit().await?;
570                } else {
571                    tx.rollback().await?;
572                    out.clear();
573                }
574            }
575            TenantExecutorInner::Conn(ref mut conn) => {
576                for (idx, body) in items.iter().enumerate() {
577                    let sp = format!("sp_{}", idx);
578                    sqlx::query(&format!("SAVEPOINT {}", sp))
579                        .execute(&mut **conn)
580                        .await?;
581                    let include_pk = body.contains_key(&entity.pk_columns[0]);
582                    let q = insert(
583                        entity,
584                        body,
585                        include_pk,
586                        schema_override,
587                        rls_tenant_id,
588                        caller_user_id,
589                        dialect,
590                    );
591                    match Self::execute_returning_one_conn(conn, &q).await {
592                        Ok(row) => {
593                            sqlx::query(&format!("RELEASE SAVEPOINT {}", sp))
594                                .execute(&mut **conn)
595                                .await?;
596                            out.push(row.unwrap_or(Value::Null));
597                        }
598                        Err(e) => {
599                            sqlx::query(&format!("ROLLBACK TO SAVEPOINT {}", sp))
600                                .execute(&mut **conn)
601                                .await?;
602                            row_errors.push((idx, e));
603                        }
604                    }
605                }
606                if !row_errors.is_empty() {
607                    out.clear();
608                }
609            }
610        }
611        Ok((out, row_errors))
612    }
613
614    /// 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.
615    /// When caller_user_id is Some, updated_by is set on each row.
616    pub async fn bulk_update<'a>(
617        executor: &mut TenantExecutor<'a>,
618        entity: &ResolvedEntity,
619        items: &[HashMap<String, Value>],
620        schema_override: Option<&str>,
621        caller_user_id: Option<&str>,
622        dialect: &dyn Dialect,
623    ) -> Result<Vec<Value>, AppError> {
624        let bulk_limit = bulk_limit();
625        if items.len() > bulk_limit {
626            return Err(AppError::BadRequest(format!(
627                "bulk update limited to {} items",
628                bulk_limit
629            )));
630        }
631        let pk = &entity.pk_columns[0];
632        let mut out = Vec::with_capacity(items.len());
633        match executor.executor {
634            TenantExecutorInner::Pool(pool) => {
635                let mut tx = pool.begin().await?;
636                for body in items {
637                    let id = body.get(pk).ok_or_else(|| {
638                        AppError::Validation(format!("each item must have '{}'", pk))
639                    })?;
640                    let mut body_without_pk = body.clone();
641                    body_without_pk.remove(pk);
642                    let q = update(
643                        entity,
644                        id,
645                        &body_without_pk,
646                        schema_override,
647                        caller_user_id,
648                        dialect,
649                    );
650                    if let Some(row) = Self::execute_returning_one_tx(&mut tx, &q).await? {
651                        out.push(row);
652                    }
653                }
654                tx.commit().await?;
655            }
656            TenantExecutorInner::Conn(ref mut conn) => {
657                for body in items {
658                    let id = body.get(pk).ok_or_else(|| {
659                        AppError::Validation(format!("each item must have '{}'", pk))
660                    })?;
661                    let mut body_without_pk = body.clone();
662                    body_without_pk.remove(pk);
663                    let q = update(
664                        entity,
665                        id,
666                        &body_without_pk,
667                        schema_override,
668                        caller_user_id,
669                        dialect,
670                    );
671                    if let Some(row) = Self::execute_returning_one_conn(conn, &q).await? {
672                        out.push(row);
673                    }
674                }
675            }
676        }
677        Ok(out)
678    }
679
680    /// Like `bulk_update` but uses savepoints to isolate per-row DB errors.
681    /// Missing pk on an item is recorded as a row error rather than aborting early.
682    /// Returns `(successful_rows, row_errors)`. If any errors occur the transaction is
683    /// rolled back and successful_rows will be empty.
684    pub async fn bulk_update_collecting<'a>(
685        executor: &mut TenantExecutor<'a>,
686        entity: &ResolvedEntity,
687        items: &[HashMap<String, Value>],
688        schema_override: Option<&str>,
689        caller_user_id: Option<&str>,
690        dialect: &dyn Dialect,
691    ) -> Result<(Vec<Value>, Vec<(usize, AppError)>), AppError> {
692        let bulk_limit = bulk_limit();
693        if items.len() > bulk_limit {
694            return Err(AppError::BadRequest(format!(
695                "bulk update limited to {} items",
696                bulk_limit
697            )));
698        }
699        let pk = entity.pk_columns[0].clone();
700        let mut out = Vec::with_capacity(items.len());
701        let mut row_errors: Vec<(usize, AppError)> = Vec::new();
702        match executor.executor {
703            TenantExecutorInner::Pool(pool) => {
704                let mut tx = pool.begin().await?;
705                for (idx, body) in items.iter().enumerate() {
706                    let id = match body.get(&pk) {
707                        Some(id) => id.clone(),
708                        None => {
709                            row_errors.push((
710                                idx,
711                                AppError::Validation(format!("each item must have '{}'", pk)),
712                            ));
713                            continue;
714                        }
715                    };
716                    let sp = format!("sp_{}", idx);
717                    sqlx::query(&format!("SAVEPOINT {}", sp))
718                        .execute(&mut *tx)
719                        .await?;
720                    let mut body_without_pk = body.clone();
721                    body_without_pk.remove(&pk);
722                    let q = update(
723                        entity,
724                        &id,
725                        &body_without_pk,
726                        schema_override,
727                        caller_user_id,
728                        dialect,
729                    );
730                    match Self::execute_returning_one_tx(&mut tx, &q).await {
731                        Ok(Some(row)) => {
732                            sqlx::query(&format!("RELEASE SAVEPOINT {}", sp))
733                                .execute(&mut *tx)
734                                .await?;
735                            out.push(row);
736                        }
737                        Ok(None) => {
738                            sqlx::query(&format!("RELEASE SAVEPOINT {}", sp))
739                                .execute(&mut *tx)
740                                .await?;
741                        }
742                        Err(e) => {
743                            sqlx::query(&format!("ROLLBACK TO SAVEPOINT {}", sp))
744                                .execute(&mut *tx)
745                                .await?;
746                            row_errors.push((idx, e));
747                        }
748                    }
749                }
750                if row_errors.is_empty() {
751                    tx.commit().await?;
752                } else {
753                    tx.rollback().await?;
754                    out.clear();
755                }
756            }
757            TenantExecutorInner::Conn(ref mut conn) => {
758                for (idx, body) in items.iter().enumerate() {
759                    let id = match body.get(&pk) {
760                        Some(id) => id.clone(),
761                        None => {
762                            row_errors.push((
763                                idx,
764                                AppError::Validation(format!("each item must have '{}'", pk)),
765                            ));
766                            continue;
767                        }
768                    };
769                    let sp = format!("sp_{}", idx);
770                    sqlx::query(&format!("SAVEPOINT {}", sp))
771                        .execute(&mut **conn)
772                        .await?;
773                    let mut body_without_pk = body.clone();
774                    body_without_pk.remove(&pk);
775                    let q = update(
776                        entity,
777                        &id,
778                        &body_without_pk,
779                        schema_override,
780                        caller_user_id,
781                        dialect,
782                    );
783                    match Self::execute_returning_one_conn(conn, &q).await {
784                        Ok(Some(row)) => {
785                            sqlx::query(&format!("RELEASE SAVEPOINT {}", sp))
786                                .execute(&mut **conn)
787                                .await?;
788                            out.push(row);
789                        }
790                        Ok(None) => {
791                            sqlx::query(&format!("RELEASE SAVEPOINT {}", sp))
792                                .execute(&mut **conn)
793                                .await?;
794                        }
795                        Err(e) => {
796                            sqlx::query(&format!("ROLLBACK TO SAVEPOINT {}", sp))
797                                .execute(&mut **conn)
798                                .await?;
799                            row_errors.push((idx, e));
800                        }
801                    }
802                }
803                if !row_errors.is_empty() {
804                    out.clear();
805                }
806            }
807        }
808        Ok((out, row_errors))
809    }
810
811    /// Bulk delete by id with per-row savepoint isolation, mirroring `bulk_update_collecting`.
812    /// Each id is deleted via [`Self::delete`] on the shared transaction/connection, so versioning
813    /// snapshots and audit rows are honored exactly as for single deletes. Ids that do not exist are
814    /// skipped silently (no row and no error), matching single-delete semantics.
815    /// Returns `(deleted_rows, row_errors)`. If any error occurs the transaction is rolled back and
816    /// deleted_rows is cleared (all-or-nothing).
817    pub async fn bulk_delete_collecting<'a>(
818        executor: &mut TenantExecutor<'a>,
819        entity: &ResolvedEntity,
820        ids: &[Value],
821        schema_override: Option<&str>,
822        caller_user_id: Option<&str>,
823        dialect: &dyn Dialect,
824    ) -> Result<(Vec<Value>, Vec<(usize, AppError)>), AppError> {
825        let bulk_limit = bulk_limit();
826        if ids.len() > bulk_limit {
827            return Err(AppError::BadRequest(format!(
828                "bulk delete limited to {} items",
829                bulk_limit
830            )));
831        }
832        let mut out = Vec::with_capacity(ids.len());
833        let mut row_errors: Vec<(usize, AppError)> = Vec::new();
834        match executor.executor {
835            TenantExecutorInner::Pool(pool) => {
836                let mut tx = pool.begin().await?;
837                for (idx, id) in ids.iter().enumerate() {
838                    let sp = format!("sp_{}", idx);
839                    sqlx::query(&format!("SAVEPOINT {}", sp))
840                        .execute(&mut *tx)
841                        .await?;
842                    // Delete on the shared tx connection (Conn executor) so it participates in this
843                    // transaction rather than autocommitting per row.
844                    let mut ex = TenantExecutor::conn(&mut tx, dialect);
845                    let res = Self::delete(
846                        &mut ex,
847                        entity,
848                        id,
849                        schema_override,
850                        caller_user_id,
851                        dialect,
852                    )
853                    .await;
854                    match res {
855                        Ok(Some(row)) => {
856                            sqlx::query(&format!("RELEASE SAVEPOINT {}", sp))
857                                .execute(&mut *tx)
858                                .await?;
859                            out.push(row);
860                        }
861                        Ok(None) => {
862                            sqlx::query(&format!("RELEASE SAVEPOINT {}", sp))
863                                .execute(&mut *tx)
864                                .await?;
865                        }
866                        Err(e) => {
867                            sqlx::query(&format!("ROLLBACK TO SAVEPOINT {}", sp))
868                                .execute(&mut *tx)
869                                .await?;
870                            row_errors.push((idx, e));
871                        }
872                    }
873                }
874                if row_errors.is_empty() {
875                    tx.commit().await?;
876                } else {
877                    tx.rollback().await?;
878                    out.clear();
879                }
880            }
881            TenantExecutorInner::Conn(ref mut conn) => {
882                for (idx, id) in ids.iter().enumerate() {
883                    let sp = format!("sp_{}", idx);
884                    sqlx::query(&format!("SAVEPOINT {}", sp))
885                        .execute(&mut **conn)
886                        .await?;
887                    let mut ex = TenantExecutor::conn(conn, dialect);
888                    let res = Self::delete(
889                        &mut ex,
890                        entity,
891                        id,
892                        schema_override,
893                        caller_user_id,
894                        dialect,
895                    )
896                    .await;
897                    match res {
898                        Ok(Some(row)) => {
899                            sqlx::query(&format!("RELEASE SAVEPOINT {}", sp))
900                                .execute(&mut **conn)
901                                .await?;
902                            out.push(row);
903                        }
904                        Ok(None) => {
905                            sqlx::query(&format!("RELEASE SAVEPOINT {}", sp))
906                                .execute(&mut **conn)
907                                .await?;
908                        }
909                        Err(e) => {
910                            sqlx::query(&format!("ROLLBACK TO SAVEPOINT {}", sp))
911                                .execute(&mut **conn)
912                                .await?;
913                            row_errors.push((idx, e));
914                        }
915                    }
916                }
917                if !row_errors.is_empty() {
918                    out.clear();
919                }
920            }
921        }
922        Ok((out, row_errors))
923    }
924
925    /// Execute a history SELECT that returns multiple rows (used by list_history handler).
926    /// Binds: params[0] = pk value.
927    pub async fn query_history_many<'a>(
928        executor: &mut TenantExecutor<'a>,
929        sql: &str,
930        params: &[Value],
931    ) -> Result<Vec<Value>, AppError> {
932        Self::query_many_exec(executor, sql, params).await
933    }
934
935    /// Execute a history SELECT that returns one row (used by read_history_version handler).
936    /// Binds: $1 = pk value, $2 = version (i64).
937    pub async fn query_history_one<'a>(
938        executor: &mut TenantExecutor<'a>,
939        sql: &str,
940        id: &Value,
941        version: i64,
942    ) -> Result<Option<Value>, AppError> {
943        tracing::debug!(sql = %sql, "history query");
944        let mut query = sqlx::query(sql);
945        query = query.bind(Self::to_sqlx_param(id));
946        query = query.bind(version);
947        let row = match executor.executor {
948            TenantExecutorInner::Pool(pool) => query.fetch_optional(pool).await?,
949            TenantExecutorInner::Conn(ref mut conn) => query.fetch_optional(&mut **conn).await?,
950        };
951        Ok(row.map(|r| row_to_json(&r)))
952    }
953
954    async fn query_one_exec<'a>(
955        executor: &mut TenantExecutor<'a>,
956        sql: &str,
957        params: &[Value],
958    ) -> Result<Option<Value>, AppError> {
959        tracing::debug!(sql = %sql, params = ?params, "query");
960        let bind = Self::to_sqlx_param(&params[0]);
961        let row = match executor.executor {
962            TenantExecutorInner::Pool(pool) => {
963                sqlx::query(sql).bind(bind).fetch_optional(pool).await?
964            }
965            TenantExecutorInner::Conn(ref mut conn) => {
966                sqlx::query(sql)
967                    .bind(bind)
968                    .fetch_optional(&mut **conn)
969                    .await?
970            }
971        };
972        Ok(row.map(|r| row_to_json(&r)))
973    }
974
975    async fn query_many_exec<'a>(
976        executor: &mut TenantExecutor<'a>,
977        sql: &str,
978        params: &[Value],
979    ) -> Result<Vec<Value>, AppError> {
980        tracing::debug!(sql = %sql, params = ?params, "query");
981        let mut query = sqlx::query(sql);
982        for p in params {
983            query = query.bind(Self::to_sqlx_param(p));
984        }
985        let rows = match executor.executor {
986            TenantExecutorInner::Pool(pool) => query.fetch_all(pool).await?,
987            TenantExecutorInner::Conn(ref mut conn) => query.fetch_all(&mut **conn).await?,
988        };
989        Ok(rows.iter().map(row_to_json).collect())
990    }
991
992    async fn execute_returning_one_exec<'a>(
993        executor: &mut TenantExecutor<'a>,
994        q: &QueryBuf,
995    ) -> Result<Option<Value>, AppError> {
996        tracing::debug!(sql = %q.sql, params = ?q.params, "query");
997        let mut query = sqlx::query(&q.sql);
998        for p in &q.params {
999            query = query.bind(Self::to_sqlx_param(p));
1000        }
1001        let row = match executor.executor {
1002            TenantExecutorInner::Pool(pool) => query.fetch_optional(pool).await?,
1003            TenantExecutorInner::Conn(ref mut conn) => query.fetch_optional(&mut **conn).await?,
1004        };
1005        Ok(row.map(|r| row_to_json(&r)))
1006    }
1007
1008    async fn execute_returning_one_with_params_exec<'a>(
1009        executor: &mut TenantExecutor<'a>,
1010        sql: &str,
1011        params: &[Value],
1012    ) -> Result<Option<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 row = match executor.executor {
1019            TenantExecutorInner::Pool(pool) => query.fetch_optional(pool).await?,
1020            TenantExecutorInner::Conn(ref mut conn) => query.fetch_optional(&mut **conn).await?,
1021        };
1022        Ok(row.map(|r| row_to_json(&r)))
1023    }
1024
1025    async fn execute_returning_one_conn(
1026        conn: &mut Connection,
1027        q: &QueryBuf,
1028    ) -> Result<Option<Value>, AppError> {
1029        tracing::debug!(sql = %q.sql, params = ?q.params, "query (conn)");
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 = query.fetch_optional(conn).await?;
1035        Ok(row.map(|r| row_to_json(&r)))
1036    }
1037
1038    async fn execute_returning_one_tx(
1039        tx: &mut Connection,
1040        q: &QueryBuf,
1041    ) -> Result<Option<Value>, AppError> {
1042        tracing::debug!(sql = %q.sql, params = ?q.params, "query (tx)");
1043        let mut query = sqlx::query(&q.sql);
1044        for p in &q.params {
1045            query = query.bind(Self::to_sqlx_param(p));
1046        }
1047        let row = query.fetch_optional(&mut *tx).await?;
1048        Ok(row.map(|r| row_to_json(&r)))
1049    }
1050
1051    fn to_sqlx_param(v: &Value) -> BindValue {
1052        BindValue::from_json(v).unwrap_or(BindValue::Null)
1053    }
1054
1055    /// Snapshot + UPDATE in one transaction (versioning path for update).
1056    async fn run_versioned_update<'a>(
1057        executor: &mut TenantExecutor<'a>,
1058        id: &Value,
1059        snap_q: QueryBuf,
1060        upd_q: QueryBuf,
1061        prune_q: Option<QueryBuf>,
1062        keep_versions: Option<i64>,
1063    ) -> Result<Option<Value>, AppError> {
1064        match executor.executor {
1065            TenantExecutorInner::Pool(pool) => {
1066                let mut tx = pool.begin().await?;
1067                // Snapshot (INSERT INTO _history SELECT ...)
1068                let mut snap = sqlx::query(&snap_q.sql);
1069                snap = snap.bind(Self::to_sqlx_param(&snap_q.params[0])); // operation
1070                snap = snap.bind(Self::to_sqlx_param(id)); // pk
1071                snap.execute(&mut *tx).await?;
1072                // Update
1073                let mut upd = sqlx::query(&upd_q.sql);
1074                for p in &upd_q.params {
1075                    upd = upd.bind(Self::to_sqlx_param(p));
1076                }
1077                let row = upd.fetch_optional(&mut *tx).await?.map(|r| row_to_json(&r));
1078                // Prune
1079                if let (Some(pq), Some(kv)) = (prune_q, keep_versions) {
1080                    let mut pr = sqlx::query(&pq.sql);
1081                    pr = pr.bind(Self::to_sqlx_param(id));
1082                    pr = pr.bind(kv);
1083                    pr.execute(&mut *tx).await?;
1084                }
1085                tx.commit().await?;
1086                Ok(row)
1087            }
1088            TenantExecutorInner::Conn(ref mut conn) => {
1089                // On an RLS connection we can't open a nested transaction; use SAVEPOINT.
1090                sqlx::query("SAVEPOINT sp_versioned_update")
1091                    .execute(&mut **conn)
1092                    .await?;
1093                let snap_res = async {
1094                    let mut snap = sqlx::query(&snap_q.sql);
1095                    snap = snap.bind(Self::to_sqlx_param(&snap_q.params[0]));
1096                    snap = snap.bind(Self::to_sqlx_param(id));
1097                    snap.execute(&mut **conn).await?;
1098                    let mut upd = sqlx::query(&upd_q.sql);
1099                    for p in &upd_q.params {
1100                        upd = upd.bind(Self::to_sqlx_param(p));
1101                    }
1102                    let row = upd
1103                        .fetch_optional(&mut **conn)
1104                        .await?
1105                        .map(|r| row_to_json(&r));
1106                    if let (Some(pq), Some(kv)) = (prune_q, keep_versions) {
1107                        let mut pr = sqlx::query(&pq.sql);
1108                        pr = pr.bind(Self::to_sqlx_param(id));
1109                        pr = pr.bind(kv);
1110                        pr.execute(&mut **conn).await?;
1111                    }
1112                    Ok::<_, sqlx::Error>(row)
1113                }
1114                .await;
1115                match snap_res {
1116                    Ok(row) => {
1117                        sqlx::query("RELEASE SAVEPOINT sp_versioned_update")
1118                            .execute(&mut **conn)
1119                            .await?;
1120                        Ok(row)
1121                    }
1122                    Err(e) => {
1123                        sqlx::query("ROLLBACK TO SAVEPOINT sp_versioned_update")
1124                            .execute(&mut **conn)
1125                            .await?;
1126                        Err(AppError::Db(e))
1127                    }
1128                }
1129            }
1130        }
1131    }
1132
1133    /// Snapshot + DELETE in one transaction (versioning path for delete).
1134    async fn run_versioned_delete<'a>(
1135        executor: &mut TenantExecutor<'a>,
1136        id: &Value,
1137        snap_q: QueryBuf,
1138        del_q: QueryBuf,
1139    ) -> Result<Option<Value>, AppError> {
1140        match executor.executor {
1141            TenantExecutorInner::Pool(pool) => {
1142                let mut tx = pool.begin().await?;
1143                let mut snap = sqlx::query(&snap_q.sql);
1144                snap = snap.bind(Self::to_sqlx_param(&snap_q.params[0])); // operation
1145                snap = snap.bind(Self::to_sqlx_param(id)); // pk
1146                snap.execute(&mut *tx).await?;
1147                let mut del = sqlx::query(&del_q.sql);
1148                del = del.bind(Self::to_sqlx_param(id));
1149                let row = del.fetch_optional(&mut *tx).await?.map(|r| row_to_json(&r));
1150                tx.commit().await?;
1151                Ok(row)
1152            }
1153            TenantExecutorInner::Conn(ref mut conn) => {
1154                sqlx::query("SAVEPOINT sp_versioned_delete")
1155                    .execute(&mut **conn)
1156                    .await?;
1157                let snap_res = async {
1158                    let mut snap = sqlx::query(&snap_q.sql);
1159                    snap = snap.bind(Self::to_sqlx_param(&snap_q.params[0]));
1160                    snap = snap.bind(Self::to_sqlx_param(id));
1161                    snap.execute(&mut **conn).await?;
1162                    let mut del = sqlx::query(&del_q.sql);
1163                    del = del.bind(Self::to_sqlx_param(id));
1164                    let row = del
1165                        .fetch_optional(&mut **conn)
1166                        .await?
1167                        .map(|r| row_to_json(&r));
1168                    Ok::<_, sqlx::Error>(row)
1169                }
1170                .await;
1171                match snap_res {
1172                    Ok(row) => {
1173                        sqlx::query("RELEASE SAVEPOINT sp_versioned_delete")
1174                            .execute(&mut **conn)
1175                            .await?;
1176                        Ok(row)
1177                    }
1178                    Err(e) => {
1179                        sqlx::query("ROLLBACK TO SAVEPOINT sp_versioned_delete")
1180                            .execute(&mut **conn)
1181                            .await?;
1182                        Err(AppError::Db(e))
1183                    }
1184                }
1185            }
1186        }
1187    }
1188
1189    async fn insert_audit<'a>(
1190        executor: &mut TenantExecutor<'a>,
1191        entity: &ResolvedEntity,
1192        action: &str,
1193        row: &Value,
1194        pre_row: Option<&Value>,
1195        audit_by: Option<&str>,
1196        schema_override: Option<&str>,
1197    ) -> Result<(), AppError> {
1198        let schema = schema_override.unwrap_or(&entity.schema_name);
1199        let audit_table = format!(
1200            "\"{}\".\"{}\"",
1201            schema.replace('"', "\"\""),
1202            format!("{}_audit", entity.table_name).replace('"', "\"\"")
1203        );
1204
1205        let changed = if action == "update" {
1206            pre_row.map(|pre| compute_changed_fields(pre, row, entity))
1207        } else {
1208            None
1209        };
1210
1211        let mut col_names: Vec<String> = vec![
1212            "\"audit_action\"".to_string(),
1213            "\"audit_by\"".to_string(),
1214            "\"changed_fields\"".to_string(),
1215        ];
1216        let mut placeholders: Vec<String> = Vec::new();
1217        let mut params: Vec<Value> = Vec::new();
1218
1219        params.push(Value::String(action.to_string()));
1220        placeholders.push(format!("${}", params.len()));
1221
1222        params.push(
1223            audit_by
1224                .map(|s| Value::String(s.to_string()))
1225                .unwrap_or(Value::Null),
1226        );
1227        placeholders.push(format!("${}", params.len()));
1228
1229        params.push(changed.unwrap_or(Value::Null));
1230        placeholders.push(format!("${}::jsonb", params.len()));
1231
1232        let row_obj = row.as_object();
1233        for col in &entity.columns {
1234            let raw = row_obj
1235                .and_then(|o| o.get(&col.name))
1236                .cloned()
1237                .unwrap_or(Value::Null);
1238            let val = coerce_json_value_for_pg_array(raw, col.pg_type.as_deref());
1239            let param_num = params.len() + 1;
1240            let ph = col
1241                .pg_type
1242                .as_deref()
1243                .map(|t| format!("${}::{}", param_num, t))
1244                .unwrap_or_else(|| format!("${}", param_num));
1245            col_names.push(format!("\"{}\"", col.name));
1246            placeholders.push(ph);
1247            params.push(val);
1248        }
1249
1250        let sql = format!(
1251            "INSERT INTO {} ({}) VALUES ({})",
1252            audit_table,
1253            col_names.join(", "),
1254            placeholders.join(", ")
1255        );
1256        tracing::debug!(sql = %sql, "audit insert");
1257
1258        let mut query = sqlx::query(&sql);
1259        for p in &params {
1260            query = query.bind(Self::to_sqlx_param(p));
1261        }
1262        match executor.executor {
1263            TenantExecutorInner::Pool(pool) => {
1264                query.execute(pool).await?;
1265            }
1266            TenantExecutorInner::Conn(ref mut conn) => {
1267                query.execute(&mut **conn).await?;
1268            }
1269        }
1270        Ok(())
1271    }
1272}
1273
1274fn compute_changed_fields(pre: &Value, post: &Value, entity: &ResolvedEntity) -> Value {
1275    let pre_obj = match pre.as_object() {
1276        Some(o) => o,
1277        None => return Value::Null,
1278    };
1279    let post_obj = match post.as_object() {
1280        Some(o) => o,
1281        None => return Value::Null,
1282    };
1283    let mut changes = serde_json::Map::new();
1284    for col in &entity.columns {
1285        let pre_val = pre_obj.get(&col.name).unwrap_or(&Value::Null);
1286        let post_val = post_obj.get(&col.name).unwrap_or(&Value::Null);
1287        if pre_val != post_val {
1288            let mut diff = serde_json::Map::new();
1289            diff.insert("old".to_string(), pre_val.clone());
1290            diff.insert("new".to_string(), post_val.clone());
1291            changes.insert(col.name.clone(), Value::Object(diff));
1292        }
1293    }
1294    Value::Object(changes)
1295}
1296
1297fn row_to_json(row: &DbRow) -> Value {
1298    use sqlx::Column;
1299    use sqlx::Row;
1300    let mut map = serde_json::Map::new();
1301    for col in row.columns() {
1302        let name = col.name();
1303        let v = cell_to_value(row, name);
1304        map.insert(name.to_string(), v);
1305    }
1306    Value::Object(map)
1307}
1308
1309fn cell_to_value(row: &DbRow, name: &str) -> Value {
1310    use sqlx::Row;
1311    if let Ok(Some(n)) = row.try_get::<Option<i16>, _>(name) {
1312        return Value::Number(n.into());
1313    }
1314    if let Ok(Some(n)) = row.try_get::<Option<i32>, _>(name) {
1315        return Value::Number(n.into());
1316    }
1317    if let Ok(Some(n)) = row.try_get::<Option<i64>, _>(name) {
1318        return Value::Number(n.into());
1319    }
1320    if let Ok(Some(n)) = row.try_get::<Option<f32>, _>(name) {
1321        if let Some(n) = serde_json::Number::from_f64(n as f64) {
1322            return Value::Number(n);
1323        }
1324    }
1325    if let Ok(Some(n)) = row.try_get::<Option<f64>, _>(name) {
1326        if let Some(n) = serde_json::Number::from_f64(n) {
1327            return Value::Number(n);
1328        }
1329    }
1330    if let Ok(Some(b)) = row.try_get::<Option<bool>, _>(name) {
1331        return Value::Bool(b);
1332    }
1333    #[cfg(feature = "postgres")]
1334    if let Ok(Some(vec)) = row.try_get::<Option<Vec<String>>, _>(name) {
1335        return Value::Array(vec.into_iter().map(Value::String).collect());
1336    }
1337    #[cfg(feature = "postgres")]
1338    if let Ok(Some(vec)) = row.try_get::<Option<Vec<uuid::Uuid>>, _>(name) {
1339        return Value::Array(
1340            vec.into_iter()
1341                .map(|u| Value::String(u.to_string()))
1342                .collect(),
1343        );
1344    }
1345    #[cfg(feature = "postgres")]
1346    if let Ok(Some(vec)) = row.try_get::<Option<Vec<i64>>, _>(name) {
1347        return Value::Array(vec.into_iter().map(|n| Value::Number(n.into())).collect());
1348    }
1349    if let Ok(Some(u)) = row.try_get::<Option<uuid::Uuid>, _>(name) {
1350        return Value::String(u.to_string());
1351    }
1352    if let Ok(Some(d)) = row.try_get::<Option<chrono::DateTime<chrono::Utc>>, _>(name) {
1353        return Value::String(d.to_rfc3339());
1354    }
1355    if let Ok(Some(d)) = row.try_get::<Option<chrono::NaiveDateTime>, _>(name) {
1356        return Value::String(d.format("%Y-%m-%dT%H:%M:%S%.f").to_string());
1357    }
1358    if let Ok(Some(d)) = row.try_get::<Option<chrono::NaiveDate>, _>(name) {
1359        return Value::String(d.format("%Y-%m-%d").to_string());
1360    }
1361    if let Ok(Some(s)) = row.try_get::<Option<String>, _>(name) {
1362        // Numeric columns are selected as ::text; parse so we return a JSON number not string
1363        if let Ok(n) = s.trim().parse::<f64>() {
1364            if let Some(num) = serde_json::Number::from_f64(n) {
1365                return Value::Number(num);
1366            }
1367        }
1368        return Value::String(s);
1369    }
1370    if let Ok(Some(j)) = row.try_get::<Option<serde_json::Value>, _>(name) {
1371        return j;
1372    }
1373    Value::Null
1374}
architect_sdk/service/crud.rs

architect_sdk/service/
crud.rs
