Skip to main content

backbone_orm/
generic_repository.rs

1//! `GenericCrudRepository<T, DeleteMode>` — one implementation for all entities.
2//!
3//! The generated repository file previously duplicated ~20 identical method bodies
4//! for every entity.  This module provides those implementations once so that
5//! the generator can emit a thin newtype wrapper instead of hundreds of lines of
6//! copy-paste.
7//!
8//! # Delete mode markers
9//!
10//! | Marker | Behaviour |
11//! |--------|-----------|
12//! | [`SoftDelete`] | `deleted_at` stored in `metadata` JSONB; active rows filter `IS NULL` |
13//! | [`HardDelete`] | Rows are permanently deleted; no trash concept |
14//!
15//! # Generated output (after this change)
16//!
17//! ```rust,ignore
18//! // ~20 lines total instead of ~460-560
19//! pub struct UserRepository(
20//!     backbone_orm::GenericCrudRepository<User, backbone_orm::SoftDelete>
21//! );
22//! impl std::ops::Deref for UserRepository {
23//!     type Target = backbone_orm::GenericCrudRepository<User, backbone_orm::SoftDelete>;
24//!     fn deref(&self) -> &Self::Target { &self.0 }
25//! }
26//! impl UserRepository {
27//!     pub fn new(pool: sqlx::PgPool) -> Self {
28//!         Self(backbone_orm::GenericCrudRepository::new(pool, "users"))
29//!     }
30//!     // entity-specific methods only:
31//!     pub async fn find_by_email(&self, email: &str) -> anyhow::Result<Option<User>> { … }
32//!     pub async fn partial_update(…) { … }
33//!     pub async fn list_paginated_filtered(…) { … }
34//! }
35//! backbone_core::impl_crud_repository!(UserRepository, User, soft_delete);
36//! ```
37
38use std::collections::HashMap;
39use std::marker::PhantomData;
40
41use anyhow::Result;
42use serde::Serialize;
43use sqlx::postgres::PgRow;
44use sqlx::{FromRow, PgPool};
45use uuid::Uuid;
46
47use crate::repository::{
48    DatabaseOperations, PaginatedResult, PaginationInfo, PaginationParams, PostgresRepository,
49};
50
51// ─── EntityRepoMeta trait ─────────────────────────────────────────────────────
52
53/// Metadata trait that entities implement so `GenericCrudRepository` can provide
54/// `list_paginated_filtered` and `list_deleted_filtered` without per-entity boilerplate.
55///
56/// Generated once per entity by the `rust` generator.
57///
58/// | Method | Purpose |
59/// |--------|---------|
60/// | `column_types` | PostgreSQL cast hints (e.g. `{"id": "uuid"}`) |
61/// | `search_fields` | Text columns searched with `ILIKE` |
62pub trait EntityRepoMeta {
63    /// PostgreSQL type hints for filter/sort (e.g. `{"id": "uuid"}`).
64    fn column_types() -> HashMap<String, String>;
65    /// Text columns to include in full-text / ILIKE search.
66    fn search_fields() -> &'static [&'static str];
67
68    // ── Field-level security (response shaping) ──
69    // Names below are RESPONSE JSON keys (camelCase), not DB columns — they are
70    // matched against the serialized response object. Defaults = no restriction.
71
72    /// Response fields visible only to the row owner and platform/root callers
73    /// (`@private`). Stripped for everyone else.
74    fn private_fields() -> &'static [&'static str] {
75        &[]
76    }
77
78    /// Response fields (camelCase) that never leave the server (`@sensitive`): no
79    /// caller is served them, a platform caller or the row's owner included.
80    fn secret_fields() -> &'static [&'static str] {
81        &[]
82    }
83
84    /// The `@sensitive` fields (camelCase response keys) of the model a to-one
85    /// relation points at, stripped from the related row an `?include=` expands.
86    /// Keyed by the relation name of [`relations`](Self::relations). Default: none.
87    fn relation_secret_fields(_relation: &str) -> &'static [&'static str] {
88        &[]
89    }
90
91    /// The response field (camelCase) holding the owner/company id (`@owner`),
92    /// compared against the caller's access scope to decide private-field
93    /// visibility. `None` = the entity has no owner concept.
94    fn owner_field() -> Option<&'static str> {
95        None
96    }
97
98    // ── Row-level company isolation ──
99
100    /// The DB column that scopes rows to a company (snake_case, e.g. `company_id`).
101    /// `None` = the entity is global (reference data) and is not fenced.
102    ///
103    /// The generator emits this **structurally** — from the presence of the company
104    /// column, a fact it cannot forget — and an entity opts OUT with `@global`.
105    /// The polarity is deliberate: `owner_field()` above is opt-IN via `@owner`, and
106    /// in a year of generated entities not one model ever declared it, so the whole
107    /// field-security path silently never ran. Security that must be remembered, in a
108    /// file nobody edits for security reasons, defaults to open and stays open. Here
109    /// silence means fenced, and forgetting is the safe direction.
110    ///
111    /// This is a *row-visibility fence*, the same category as `SoftDelete`'s
112    /// unconditional `deleted_at IS NULL` — not domain logic.
113    fn company_field() -> Option<&'static str> {
114        None
115    }
116
117    /// To-one relations expandable via `?include=<name>`, as
118    /// `(relation_name, target_table, local_fk_field)` — all response keys
119    /// (relation_name/local_fk in camelCase; target_table is the DB table).
120    /// Default: none.
121    fn relations() -> &'static [(&'static str, &'static str, &'static str)] {
122        &[]
123    }
124}
125
126// ─── Delete-mode markers ──────────────────────────────────────────────────────
127
128// ─── Row-level company fence ───────────────────────────────────────────────────
129
130/// The caller could not be scoped to a company for an entity that requires one.
131///
132/// Deliberately an error and not an empty result: an empty list is a silent lie that
133/// looks identical to "no data" and gets debugged for days, whereas this names the
134/// wiring bug. Callers should surface it as 403.
135#[derive(Debug, thiserror::Error)]
136#[error("entity is company-scoped ({column}) but the request carries no company scope")]
137pub struct MissingCompanyScope {
138    /// The column that would have fenced the query.
139    pub column: &'static str,
140}
141
142/// Build the SQL fence for `T`, or fail closed.
143///
144/// - entity is global (`company_field() == None`) → no fence, reference data reads freely
145/// - entity is fenced but the request has no company → **`Err`**, never an unfenced query
146/// - both present → `col = '<uuid>'`, ANDed into the WHERE by the caller
147///
148/// The literal is safe to interpolate: `column` is generator-emitted (never client
149/// input) and `Uuid`'s `Display` can only produce hex and dashes, so neither side can
150/// carry a quote. A base condition is code-authored SQL with no parameter slot, the
151/// same seam the soft-delete guard uses.
152pub fn company_fence<T: EntityRepoMeta>(company: Option<Uuid>) -> Result<Option<String>, MissingCompanyScope> {
153    match (T::company_field(), company) {
154        (None, _) => Ok(None),
155        (Some(column), None) => Err(MissingCompanyScope { column }),
156        (Some(column), Some(id)) => Ok(Some(format!("{column} = '{id}'"))),
157    }
158}
159
160/// Remove any caller-supplied filter on the company column.
161///
162/// The HTTP filter map is `#[serde(flatten)]`, so `?company_id=<victim>` arrives as an
163/// ordinary filter and the builder honours it. The fence is ANDed, so a client cannot
164/// widen past its own company — but leaving the key in lets a caller AND a *second*,
165/// contradictory company predicate and turn every list into an empty-vs-nonempty oracle
166/// for "does company X have rows matching Y". Strip it: a client may narrow **within**
167/// its company, never speak about the company at all.
168///
169/// Handles the snake_case column, its camelCase response key, and both operator forms
170/// (`company_id[eq]=…`).
171pub fn strip_client_company_filters<T: EntityRepoMeta>(filters: &mut HashMap<String, String>) {
172    let Some(column) = T::company_field() else {
173        return;
174    };
175    let camel = snake_to_camel(column);
176    filters.retain(|key, _| {
177        let base = key.split('[').next().unwrap_or(key);
178        !base.eq_ignore_ascii_case(column) && !base.eq_ignore_ascii_case(&camel)
179    });
180}
181
182/// AND two optional SQL conditions into the single base-condition argument the filtered
183/// queries take, so the soft-delete guard and the company fence travel as one string.
184pub fn and_conditions(a: Option<&str>, b: Option<String>) -> Option<String> {
185    match (a, b) {
186        (None, None) => None,
187        (Some(a), None) => Some(a.to_string()),
188        (None, Some(b)) => Some(b),
189        (Some(a), Some(b)) => Some(format!("{a} AND {b}")),
190    }
191}
192
193fn snake_to_camel(s: &str) -> String {
194    let mut out = String::with_capacity(s.len());
195    let mut upper = false;
196    for c in s.chars() {
197        if c == '_' {
198            upper = true;
199        } else if upper {
200            out.push(c.to_ascii_uppercase());
201            upper = false;
202        } else {
203            out.push(c);
204        }
205    }
206    out
207}
208
209/// The not-deleted predicate an update of a soft-delete entity carries, against the alias `t`.
210const ACTIVE_ROW_GUARD: &str = " AND t.metadata->>'deleted_at' IS NULL";
211
212/// Marker: entity uses JSONB-based soft delete (`metadata->>'deleted_at'`).
213///
214/// Active rows satisfy `metadata->>'deleted_at' IS NULL`.
215pub struct SoftDelete;
216
217/// Marker: entity uses hard deletes only — no soft delete / trash concept.
218pub struct HardDelete;
219
220// ─── GenericCrudRepository ────────────────────────────────────────────────────
221
222/// Generic PostgreSQL CRUD repository parameterised by entity type `T` and
223/// delete strategy `D` ([`SoftDelete`] or [`HardDelete`]).
224///
225/// All standard CRUD methods are implemented here once.  Generated repository
226/// structs wrap this type via newtype + `Deref` and only add entity-specific
227/// methods (`find_by_*`, `exists_by_*`, `partial_update`, filtered listings).
228pub struct GenericCrudRepository<T, D = SoftDelete>
229where
230    T: for<'r> FromRow<'r, PgRow> + Send + Unpin,
231{
232    inner: PostgresRepository<T>,
233    _mode: PhantomData<D>,
234}
235
236// ─── Common (both modes) ──────────────────────────────────────────────────────
237
238impl<T, D> GenericCrudRepository<T, D>
239where
240    T: for<'r> FromRow<'r, PgRow> + Send + Unpin,
241{
242    pub fn new(pool: PgPool, table_name: &str) -> Self {
243        Self {
244            inner: PostgresRepository::new(pool, table_name),
245            _mode: PhantomData,
246        }
247    }
248
249    pub fn pool(&self) -> &PgPool {
250        self.inner.pool()
251    }
252
253    pub fn table_name(&self) -> &str {
254        self.inner.table_name()
255    }
256
257    /// Insert a new entity row and return the created row.
258    pub async fn create(&self, entity: &T) -> Result<T>
259    where
260        T: Serialize + Send + Sync,
261    {
262        self.inner.create(entity).await
263    }
264
265    /// Insert multiple entity rows inside a single transaction: all of them, or none.
266    ///
267    /// The first row the database refuses rolls the whole batch back, and the error is a
268    /// [`BatchItemError`] naming that row's position in `entities`. The transaction runs on the
269    /// request's own connection when a request scope holds one, so it carries the same fence
270    /// (company, org units, audit attribution) as a single create.
271    pub async fn bulk_create(&self, entities: &[T]) -> Result<Vec<T>>
272    where
273        T: Serialize + Send + Sync,
274    {
275        if entities.is_empty() {
276            return Ok(Vec::new());
277        }
278        let table = self.table_name();
279        let unit = crate::write_tx::begin_unit(self.pool(), true).await?;
280        let rows = async {
281            let mut conn = unit.conn().await;
282            insert_each(&mut conn, table, entities).await
283        }
284        .await;
285        unit.finish(rows).await
286    }
287
288    /// Update many rows from `(before, after)` pairs inside one transaction, writing for each
289    /// only the columns that differ — the batch twin of a single changed-columns update — and
290    /// only while `guard` holds (see [`crate::repository::PostgresRepository::update_guarded`]).
291    ///
292    /// A row that no longer matches rolls the whole batch back with a [`BatchItemError`]
293    /// naming its position.
294    async fn bulk_update_changed_guarded(&self, pairs: &[(T, T)], guard: &str) -> Result<Vec<T>>
295    where
296        T: Serialize + Send + Sync,
297    {
298        if pairs.is_empty() {
299            return Ok(Vec::new());
300        }
301        let table = self.table_name();
302        let unit = crate::write_tx::begin_unit(self.pool(), true).await?;
303        let rows = async {
304            let mut conn = unit.conn().await;
305            update_each_changed(&mut conn, table, guard, pairs).await
306        }
307        .await;
308        unit.finish(rows).await
309    }
310
311    // ── Generic unique-field lookups ──────────────────────────────────────────
312    //
313    // These are shared across both delete modes. The SQL is the same for both;
314    // the soft-delete guard is added separately by the per-mode wrappers below.
315
316    /// Internal: query by a text field with a caller-supplied extra condition.
317    async fn find_by_text_field_with_cond(
318        &self,
319        field: &str,
320        value: &str,
321        extra: &str,
322    ) -> Result<Option<T>> {
323        let query = format!(
324            "SELECT * FROM {} WHERE {} = $1{}",
325            self.table_name(), field, extra
326        );
327        let result = crate::company_scope::fetch_optional_scoped(
328            self.pool(),
329            sqlx::query_as::<_, T>(&query).bind(value),
330        )
331        .await?;
332        Ok(result)
333    }
334
335    /// Internal: existence check by a text field with a caller-supplied extra condition.
336    async fn exists_by_text_field_with_cond(
337        &self,
338        field: &str,
339        value: &str,
340        extra: &str,
341    ) -> Result<bool> {
342        let query = format!(
343            "SELECT 1 FROM {} WHERE {} = $1{} LIMIT 1",
344            self.table_name(), field, extra
345        );
346        let result = crate::company_scope::fetch_optional_scalar_scoped(
347            self.pool(),
348            sqlx::query_scalar::<_, i32>(&query).bind(value),
349        )
350        .await?;
351        Ok(result.is_some())
352    }
353
354    /// Internal: query by a UUID field with a caller-supplied extra condition.
355    async fn find_by_uuid_field_with_cond(
356        &self,
357        field: &str,
358        value: Uuid,
359        extra: &str,
360    ) -> Result<Option<T>> {
361        let query = format!(
362            "SELECT * FROM {} WHERE {} = $1{}",
363            self.table_name(), field, extra
364        );
365        let result = crate::company_scope::fetch_optional_scoped(
366            self.pool(),
367            sqlx::query_as::<_, T>(&query).bind(value),
368        )
369        .await?;
370        Ok(result)
371    }
372
373    /// Internal: existence check by a UUID field with a caller-supplied extra condition.
374    async fn exists_by_uuid_field_with_cond(
375        &self,
376        field: &str,
377        value: Uuid,
378        extra: &str,
379    ) -> Result<bool> {
380        let query = format!(
381            "SELECT 1 FROM {} WHERE {} = $1{} LIMIT 1",
382            self.table_name(), field, extra
383        );
384        let result = crate::company_scope::fetch_optional_scalar_scoped(
385            self.pool(),
386            sqlx::query_scalar::<_, i32>(&query).bind(value),
387        )
388        .await?;
389        Ok(result.is_some())
390    }
391
392    /// Execute a filtered / paginated query against this entity's table.
393    ///
394    /// `base_condition` — when `Some`, raw SQL ANDed as its own parenthesised group ahead of
395    /// the caller-supplied filters, which can therefore never widen past it (not even with
396    /// an `[or]` filter). It must be code-authored; the filters map cannot carry one, since
397    /// the parser ignores every `__`-prefixed key.
398    ///
399    /// NOTE: prefer the `*_scoped` entry points for anything a client can reach —
400    /// they compose the company fence into this condition. This raw form applies no
401    /// fence and is for internal/job callers that have already established scope.
402    pub async fn run_filtered_query(
403        &self,
404        pagination: PaginationParams,
405        base_condition: Option<&str>,
406        filters: &HashMap<String, String>,
407        column_types: &HashMap<String, String>,
408        search_fields: &[&str],
409    ) -> Result<PaginatedResult<T>>
410    where
411        T: Send + Sync,
412    {
413        self.inner
414            .list_with_filters_and_base(
415                pagination,
416                base_condition,
417                filters,
418                column_types,
419                search_fields,
420            )
421            .await
422    }
423
424    /// The aggregate twin of [`Self::run_filtered_query`].
425    ///
426    /// Takes the same `base_condition` so an aggregate is computed over exactly
427    /// the row set the matching list would return — a total that counted
428    /// soft-deleted rows, or another tenant's, would be wrong in a way no
429    /// caller could see.
430    pub async fn run_aggregate_query(
431        &self,
432        spec: &crate::repository::AggregateSpec,
433        base_condition: Option<&str>,
434        filters: &HashMap<String, String>,
435        column_types: &HashMap<String, String>,
436        search_fields: &[&str],
437    ) -> Result<crate::repository::AggregateResult>
438    where
439        T: crate::EntityRepoMeta + Send + Sync,
440    {
441        // Resolve the group column's relation HERE (the only layer that
442        // knows the entity's metadata): a group on a relation FK gets its
443        // target table passed down so the inner query can carry the label.
444        let mut spec = spec.clone();
445        if let (Some(group), Some(label)) = (&spec.group_by, &spec.label_field) {
446            let _ = label;
447            let camel = snake_to_camel(group);
448            if let Some((_, table, _)) = T::relations().iter().find(|(_, _, fk)| *fk == camel) {
449                let base_fk = camel_to_snake(&camel);
450                spec.label_relation = Some((table.to_string(), base_fk));
451            }
452        }
453        self.inner
454            .aggregate_with_filters_and_base(
455                &spec,
456                base_condition,
457                filters,
458                column_types,
459                search_fields,
460            )
461            .await
462    }
463}
464
465// ─── SoftDelete mode ──────────────────────────────────────────────────────────
466
467impl<T> GenericCrudRepository<T, SoftDelete>
468where
469    T: for<'r> FromRow<'r, PgRow> + Send + Unpin,
470{
471    // ── Unique-field lookups (active records only) ────────────────────────────
472
473    /// Find an active entity by a unique text field.
474    ///
475    /// Filters `metadata->>'deleted_at' IS NULL` automatically.
476    pub async fn find_by_text_field(&self, field: &str, value: &str) -> Result<Option<T>> {
477        self.find_by_text_field_with_cond(field, value, " AND metadata->>'deleted_at' IS NULL").await
478    }
479
480    /// Check existence by a unique text field (active records only).
481    pub async fn exists_by_text_field(&self, field: &str, value: &str) -> Result<bool> {
482        self.exists_by_text_field_with_cond(field, value, " AND metadata->>'deleted_at' IS NULL").await
483    }
484
485    /// Find an active entity by a unique UUID field.
486    pub async fn find_by_uuid_field(&self, field: &str, value: Uuid) -> Result<Option<T>> {
487        self.find_by_uuid_field_with_cond(field, value, " AND metadata->>'deleted_at' IS NULL").await
488    }
489
490    /// Check existence by a unique UUID field (active records only).
491    pub async fn exists_by_uuid_field(&self, field: &str, value: Uuid) -> Result<bool> {
492        self.exists_by_uuid_field_with_cond(field, value, " AND metadata->>'deleted_at' IS NULL").await
493    }
494
495    // ── Filtered pagination (requires EntityRepoMeta on T) ────────────────────
496
497    /// Paginate active entities with filter and search support.
498    ///
499    /// Requires `T: EntityRepoMeta` for column type hints and search fields.
500    pub async fn list_paginated_filtered(
501        &self,
502        pagination: PaginationParams,
503        filters: Option<&HashMap<String, String>>,
504    ) -> Result<PaginatedResult<T>>
505    where
506        T: EntityRepoMeta + Send + Sync,
507    {
508        let filters_map = filters.cloned().unwrap_or_default();
509        let column_types = T::column_types();
510        let search_fields_owned: Vec<&'static str> = T::search_fields().iter().copied().collect();
511        self.run_filtered_query(
512            pagination,
513            Some("metadata->>'deleted_at' IS NULL"),
514            &filters_map,
515            &column_types,
516            &search_fields_owned,
517        ).await
518    }
519
520    /// Group and reduce active entities, skipping the soft-deleted.
521    pub async fn aggregate_filtered(
522        &self,
523        spec: &crate::repository::AggregateSpec,
524        filters: Option<&HashMap<String, String>>,
525    ) -> Result<crate::repository::AggregateResult>
526    where
527        T: EntityRepoMeta + Send + Sync,
528    {
529        let filters_map = filters.cloned().unwrap_or_default();
530        let column_types = T::column_types();
531        let search_fields_owned: Vec<&'static str> = T::search_fields().iter().copied().collect();
532        self.run_aggregate_query(spec, Some("metadata->>'deleted_at' IS NULL"), &filters_map, &column_types, &search_fields_owned).await
533    }
534
535    /// Paginate active entities, fenced to `company`.
536    ///
537    /// The company-scoped counterpart of [`Self::list_paginated_filtered`] and the one any
538    /// client-reachable read should use. Fails closed: a fenced entity with no company
539    /// returns [`MissingCompanyScope`] rather than every company's rows.
540    ///
541    /// The fence is ANDed into the same base condition the soft-delete guard uses, so
542    /// it lands in SQL — which is what keeps `total` honest. A post-filter in the handler
543    /// would fetch `limit` rows and return the survivors, leaving callers unable to tell
544    /// "end of data" from "filtered", and could not fix `COUNT` at all.
545    pub async fn list_paginated_filtered_scoped(
546        &self,
547        pagination: PaginationParams,
548        filters: Option<&HashMap<String, String>>,
549        company: Option<Uuid>,
550    ) -> Result<PaginatedResult<T>>
551    where
552        T: EntityRepoMeta + Send + Sync,
553    {
554        let fence = company_fence::<T>(company)?;
555        let mut filters_map = filters.cloned().unwrap_or_default();
556        strip_client_company_filters::<T>(&mut filters_map);
557        let column_types = T::column_types();
558        let search_fields_owned: Vec<&'static str> = T::search_fields().iter().copied().collect();
559        self.run_filtered_query(
560            pagination,
561            and_conditions(Some("metadata->>'deleted_at' IS NULL"), fence).as_deref(),
562            &filters_map,
563            &column_types,
564            &search_fields_owned,
565        )
566        .await
567    }
568
569    /// Paginate soft-deleted entities, fenced to `company`.
570    ///
571    /// `/trash` is the worse leak of the two: it serves rows whose owners were told the
572    /// data is gone. Same fence, same fail-closed contract.
573    pub async fn list_deleted_filtered_scoped(
574        &self,
575        pagination: PaginationParams,
576        filters: Option<&HashMap<String, String>>,
577        company: Option<Uuid>,
578    ) -> Result<PaginatedResult<T>>
579    where
580        T: EntityRepoMeta + Send + Sync,
581    {
582        let fence = company_fence::<T>(company)?;
583        let mut filters_map = filters.cloned().unwrap_or_default();
584        strip_client_company_filters::<T>(&mut filters_map);
585        let column_types = T::column_types();
586        let search_fields: Vec<&'static str> = T::search_fields().to_vec();
587        self.run_filtered_query(
588            pagination,
589            and_conditions(Some("metadata->>'deleted_at' IS NOT NULL"), fence).as_deref(),
590            &filters_map,
591            &column_types,
592            &search_fields,
593        )
594        .await
595    }
596
597    /// Paginate soft-deleted entities with filter and search support — the trash twin of
598    /// [`Self::list_paginated_filtered`].
599    pub async fn list_deleted_filtered(
600        &self,
601        pagination: PaginationParams,
602        filters: Option<&HashMap<String, String>>,
603    ) -> Result<PaginatedResult<T>>
604    where
605        T: EntityRepoMeta + Send + Sync,
606    {
607        let filters_map = filters.cloned().unwrap_or_default();
608        let column_types = T::column_types();
609        let search_fields: Vec<&'static str> = T::search_fields().to_vec();
610        self.run_filtered_query(
611            pagination,
612            Some("metadata->>'deleted_at' IS NOT NULL"),
613            &filters_map,
614            &column_types,
615            &search_fields,
616        ).await
617    }
618
619    /// Find an active (non-deleted) entity by primary key.
620    pub async fn find_by_id(&self, id: &str) -> Result<Option<T>> {
621        let query = format!(
622            "SELECT * FROM {} WHERE id = $1::uuid AND metadata->>'deleted_at' IS NULL",
623            self.table_name()
624        );
625        let result = crate::company_scope::fetch_optional_scoped(
626            self.pool(),
627            sqlx::query_as::<_, T>(&query).bind(id),
628        )
629        .await?;
630        Ok(result)
631    }
632
633    /// Return all active (non-deleted) entities.
634    pub async fn find_all(&self) -> Result<Vec<T>> {
635        let query = format!(
636            "SELECT * FROM {} WHERE metadata->>'deleted_at' IS NULL",
637            self.table_name()
638        );
639        let results = crate::company_scope::fetch_all_scoped(
640            self.pool(),
641            sqlx::query_as::<_, T>(&query),
642        )
643        .await?;
644        Ok(results)
645    }
646
647    /// Full update of an active record; `Ok(None)` when it is missing or soft-deleted.
648    ///
649    /// The not-deleted predicate is part of the UPDATE itself, so a write racing a soft
650    /// delete can never revive the row.
651    pub async fn update(&self, id: &str, entity: &T) -> Result<Option<T>>
652    where
653        T: Serialize + Send + Sync,
654    {
655        self.inner
656            .update_guarded(id, entity, None, ACTIVE_ROW_GUARD)
657            .await
658    }
659
660    /// Update an active record from `before` (the row as read) to `after`, writing only the
661    /// columns whose value differs between the two; `Ok(None)` when the row is missing or was
662    /// soft-deleted since it was read.
663    ///
664    /// The path for a read-modify-write such as a PATCH: a column the caller did not change is
665    /// left as it now stands in the database, so a concurrent write to it (a verb moving a
666    /// status) is not overwritten with the stale value read before it.
667    pub async fn update_changed(&self, id: &str, before: &T, after: &T) -> Result<Option<T>>
668    where
669        T: Serialize + Send + Sync,
670    {
671        let changed = crate::repository::changed_columns(before, after)?;
672        self.inner
673            .update_guarded(id, after, Some(&changed), ACTIVE_ROW_GUARD)
674            .await
675    }
676
677    /// Soft-delete an entity (sets `metadata.deleted_at`).
678    pub async fn delete(&self, id: &str) -> Result<bool> {
679        self.soft_delete(id).await
680    }
681
682    /// Count active (non-deleted) entities.
683    pub async fn count(&self) -> Result<u64> {
684        self.count_active().await
685    }
686
687    /// Return `true` if an active entity with the given ID exists.
688    pub async fn exists(&self, id: &str) -> Result<bool> {
689        let query = format!(
690            "SELECT 1 FROM {} WHERE id = $1::uuid AND metadata->>'deleted_at' IS NULL LIMIT 1",
691            self.table_name()
692        );
693        let result = crate::company_scope::fetch_optional_scalar_scoped(
694            self.pool(),
695            sqlx::query_scalar::<_, i32>(&query).bind(id),
696        )
697        .await?;
698        Ok(result.is_some())
699    }
700
701    /// Paginate active entities (most-recent-first by ID).
702    pub async fn list_paginated(&self, pagination: PaginationParams) -> Result<PaginatedResult<T>> {
703        let offset = pagination.offset();
704        let limit = pagination.limit();
705        let query = format!(
706            "SELECT * FROM {} WHERE metadata->>'deleted_at' IS NULL \
707             ORDER BY id DESC LIMIT $1 OFFSET $2",
708            self.table_name()
709        );
710        let data = crate::company_scope::fetch_all_scoped(
711            self.pool(),
712            sqlx::query_as::<_, T>(&query).bind(limit as i64).bind(offset as i64),
713        )
714        .await?;
715        let total = self.count_active().await?;
716        Ok(PaginatedResult {
717            data,
718            pagination: PaginationInfo::new(pagination.page, pagination.per_page, total),
719        })
720    }
721
722    // ── Soft-delete helpers ───────────────────────────────────────────────────
723
724    /// Set `metadata.deleted_at` to NOW() (soft delete).
725    pub async fn soft_delete(&self, id: &str) -> Result<bool> {
726        let query = format!(
727            "UPDATE {} SET metadata = jsonb_set(\
728               COALESCE(metadata, '{{}}'), \
729               '{{deleted_at}}', \
730               to_jsonb(NOW())\
731             ) WHERE id = $1::uuid AND (metadata->>'deleted_at') IS NULL",
732            self.table_name()
733        );
734        let result = crate::company_scope::execute_scoped(
735            self.pool(),
736            sqlx::query(&query).bind(id),
737        )
738        .await?;
739        Ok(result.rows_affected() > 0)
740    }
741
742    /// Remove `deleted_at` from metadata, restoring the entity.
743    pub async fn restore(&self, id: &str) -> Result<Option<T>> {
744        let query = format!(
745            "UPDATE {} SET metadata = metadata - 'deleted_at' \
746             WHERE id = $1::uuid AND (metadata->>'deleted_at') IS NOT NULL \
747             RETURNING *",
748            self.table_name()
749        );
750        let result = crate::company_scope::fetch_optional_scoped(
751            self.pool(),
752            sqlx::query_as::<_, T>(&query).bind(id),
753        )
754        .await?;
755        Ok(result)
756    }
757
758    /// Paginate soft-deleted entities (trash view).
759    pub async fn list_deleted(&self, pagination: PaginationParams) -> Result<PaginatedResult<T>> {
760        let offset = pagination.offset();
761        let limit = pagination.limit();
762        let query = format!(
763            "SELECT * FROM {} WHERE (metadata->>'deleted_at') IS NOT NULL \
764             ORDER BY (metadata->>'deleted_at') DESC LIMIT $1 OFFSET $2",
765            self.table_name()
766        );
767        let data = crate::company_scope::fetch_all_scoped(
768            self.pool(),
769            sqlx::query_as::<_, T>(&query).bind(limit as i64).bind(offset as i64),
770        )
771        .await?;
772        let count_query = format!(
773            "SELECT COUNT(*) FROM {} WHERE (metadata->>'deleted_at') IS NOT NULL",
774            self.table_name()
775        );
776        let total = crate::company_scope::fetch_one_scalar_scoped(
777            self.pool(),
778            sqlx::query_scalar::<_, i64>(&count_query),
779        )
780        .await? as u64;
781        Ok(PaginatedResult {
782            data,
783            pagination: PaginationInfo::new(pagination.page, pagination.per_page, total),
784        })
785    }
786
787    /// Permanently delete all soft-deleted rows (empty trash).
788    pub async fn empty_trash(&self) -> Result<u64> {
789        let query = format!(
790            "DELETE FROM {} WHERE (metadata->>'deleted_at') IS NOT NULL",
791            self.table_name()
792        );
793        let result = crate::company_scope::execute_scoped(self.pool(), sqlx::query(&query)).await?;
794        Ok(result.rows_affected())
795    }
796
797    /// Find a soft-deleted entity by primary key.
798    pub async fn find_deleted_by_id(&self, id: &str) -> Result<Option<T>> {
799        let query = format!(
800            "SELECT * FROM {} WHERE id = $1::uuid AND (metadata->>'deleted_at') IS NOT NULL",
801            self.table_name()
802        );
803        let result = crate::company_scope::fetch_optional_scoped(
804            self.pool(),
805            sqlx::query_as::<_, T>(&query).bind(id),
806        )
807        .await?;
808        Ok(result)
809    }
810
811    /// Permanently delete a soft-deleted entity by primary key.
812    pub async fn permanent_delete(&self, id: &str) -> Result<bool> {
813        let query = format!(
814            "DELETE FROM {} WHERE id = $1::uuid AND (metadata->>'deleted_at') IS NOT NULL",
815            self.table_name()
816        );
817        let result = crate::company_scope::execute_scoped(
818            self.pool(),
819            sqlx::query(&query).bind(id),
820        )
821        .await?;
822        Ok(result.rows_affected() > 0)
823    }
824
825    /// Count active (non-deleted) entities.
826    pub async fn count_active(&self) -> Result<u64> {
827        let query = format!(
828            "SELECT COUNT(*) FROM {} WHERE (metadata->>'deleted_at') IS NULL",
829            self.table_name()
830        );
831        let count = crate::company_scope::fetch_one_scalar_scoped(
832            self.pool(),
833            sqlx::query_scalar::<_, i64>(&query),
834        )
835        .await? as u64;
836        Ok(count)
837    }
838
839    /// Count soft-deleted entities.
840    pub async fn count_deleted(&self) -> Result<u64> {
841        let query = format!(
842            "SELECT COUNT(*) FROM {} WHERE (metadata->>'deleted_at') IS NOT NULL",
843            self.table_name()
844        );
845        let count = crate::company_scope::fetch_one_scalar_scoped(
846            self.pool(),
847            sqlx::query_scalar::<_, i64>(&query),
848        )
849        .await? as u64;
850        Ok(count)
851    }
852
853    // ── Atomic batch operations ───────────────────────────────────────────────
854    //
855    // Each runs inside a single transaction. For id-list operations the affected
856    // row count must equal the number of ids requested, otherwise the whole batch
857    // is rolled back — all-or-nothing semantics where a missing / already-in-the-
858    // target-state id (or a duplicate id) fails the entire request.
859
860    /// Soft-delete many active rows atomically.
861    pub async fn bulk_soft_delete(&self, ids: &[String]) -> Result<u64> {
862        if ids.is_empty() {
863            return Ok(0);
864        }
865        let placeholders = id_in_placeholders(ids.len());
866        let query = format!(
867            "UPDATE {} SET metadata = jsonb_set(\
868               COALESCE(metadata, '{{}}'), '{{deleted_at}}', to_jsonb(NOW())\
869             ) WHERE id IN ({placeholders}) AND (metadata->>'deleted_at') IS NULL",
870            self.table_name()
871        );
872        let unit = crate::write_tx::begin_unit(self.pool(), false).await?;
873        let outcome: Result<u64> = async {
874            let mut conn = unit.conn().await;
875            let mut q = sqlx::query(&query);
876            for id in ids {
877                q = q.bind(id);
878            }
879            let affected = q.execute(&mut *conn).await?.rows_affected();
880            if affected != ids.len() as u64 {
881                // Finishing with an error rolls the unit back.
882                return Err(anyhow::anyhow!(
883                    "bulk_soft_delete: {} of {} ids were not active/deletable; rolled back",
884                    ids.len() as u64 - affected,
885                    ids.len()
886                ));
887            }
888            Ok(affected)
889        }
890        .await;
891        unit.finish(outcome).await
892    }
893
894    /// Restore many soft-deleted rows atomically, returning the restored rows.
895    pub async fn bulk_restore(&self, ids: &[String]) -> Result<Vec<T>> {
896        if ids.is_empty() {
897            return Ok(Vec::new());
898        }
899        let placeholders = id_in_placeholders(ids.len());
900        let query = format!(
901            "UPDATE {} SET metadata = metadata - 'deleted_at' \
902             WHERE id IN ({placeholders}) AND (metadata->>'deleted_at') IS NOT NULL \
903             RETURNING *",
904            self.table_name()
905        );
906        let unit = crate::write_tx::begin_unit(self.pool(), false).await?;
907        let outcome: Result<Vec<T>> = async {
908            let mut conn = unit.conn().await;
909            let mut q = sqlx::query_as::<_, T>(&query);
910            for id in ids {
911                q = q.bind(id);
912            }
913            let rows = q.fetch_all(&mut *conn).await?;
914            if rows.len() != ids.len() {
915                return Err(anyhow::anyhow!(
916                    "bulk_restore: {} of {} ids were not in trash; rolled back",
917                    ids.len() - rows.len(),
918                    ids.len()
919                ));
920            }
921            Ok(rows)
922        }
923        .await;
924        unit.finish(outcome).await
925    }
926
927    /// Permanently delete many soft-deleted rows atomically.
928    pub async fn bulk_permanent_delete(&self, ids: &[String]) -> Result<u64> {
929        if ids.is_empty() {
930            return Ok(0);
931        }
932        let placeholders = id_in_placeholders(ids.len());
933        let query = format!(
934            "DELETE FROM {} WHERE id IN ({placeholders}) AND (metadata->>'deleted_at') IS NOT NULL",
935            self.table_name()
936        );
937        let unit = crate::write_tx::begin_unit(self.pool(), false).await?;
938        let outcome: Result<u64> = async {
939            let mut conn = unit.conn().await;
940            let mut q = sqlx::query(&query);
941            for id in ids {
942                q = q.bind(id);
943            }
944            let affected = q.execute(&mut *conn).await?.rows_affected();
945            if affected != ids.len() as u64 {
946                return Err(anyhow::anyhow!(
947                    "bulk_permanent_delete: {} of {} ids were not in trash; rolled back",
948                    ids.len() as u64 - affected,
949                    ids.len()
950                ));
951            }
952            Ok(affected)
953        }
954        .await;
955        unit.finish(outcome).await
956    }
957
958    /// Restore every soft-deleted row, returning the restored rows. A single
959    /// `UPDATE ... RETURNING` is atomic, and the returned rows let the service
960    /// layer emit a `Restored` event per entity.
961    pub async fn restore_all(&self) -> Result<Vec<T>> {
962        let query = format!(
963            "UPDATE {} SET metadata = metadata - 'deleted_at' \
964             WHERE (metadata->>'deleted_at') IS NOT NULL \
965             RETURNING *",
966            self.table_name()
967        );
968        let rows = crate::company_scope::fetch_all_scoped(
969            self.pool(),
970            sqlx::query_as::<_, T>(&query),
971        )
972        .await?;
973        Ok(rows)
974    }
975
976    /// Update many active rows atomically. Every entity must reference an
977    /// existing active (non-soft-deleted) row or the whole batch is rolled back.
978    pub async fn bulk_update(&self, entities: &[T]) -> Result<Vec<T>>
979    where
980        T: Serialize + Send + Sync,
981    {
982        bulk_update_rows(
983            self.pool(),
984            self.table_name(),
985            ACTIVE_ROW_GUARD,
986            entities,
987        )
988        .await
989    }
990
991    /// Update many active rows from `(before, after)` pairs atomically, writing only the
992    /// columns each update changed; a row missing or soft-deleted since it was read rolls the
993    /// whole batch back.
994    pub async fn bulk_update_changed(&self, pairs: &[(T, T)]) -> Result<Vec<T>>
995    where
996        T: Serialize + Send + Sync,
997    {
998        self.bulk_update_changed_guarded(pairs, ACTIVE_ROW_GUARD).await
999    }
1000}
1001
1002/// Build `"$1::uuid, $2::uuid, …"` for an `id IN (…)` clause of `n` bound ids.
1003fn id_in_placeholders(n: usize) -> String {
1004    (1..=n)
1005        .map(|i| format!("${i}::uuid"))
1006        .collect::<Vec<_>>()
1007        .join(", ")
1008}
1009
1010/// One item of a batch write failed, so the whole batch was rolled back.
1011///
1012/// `index` is the item's position in the batch as the caller passed it; `error` is why it
1013/// failed (a database refusal keeps its `sqlx::Error` inside, so a caller can still read the
1014/// constraint it broke).
1015#[derive(Debug)]
1016pub struct BatchItemError {
1017    /// Position of the failed item in the batch.
1018    pub index: usize,
1019    /// Why it failed.
1020    pub error: anyhow::Error,
1021}
1022
1023impl std::fmt::Display for BatchItemError {
1024    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
1025        write!(f, "item {} of the batch: {}; rolled back", self.index, self.error)
1026    }
1027}
1028
1029impl std::error::Error for BatchItemError {
1030    fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
1031        Some(self.error.as_ref())
1032    }
1033}
1034
1035/// Insert each entity on `conn` (an open transaction), stopping at the first failure.
1036async fn insert_each<T>(conn: &mut sqlx::PgConnection, table: &str, entities: &[T]) -> Result<Vec<T>>
1037where
1038    T: for<'r> FromRow<'r, PgRow> + Serialize + Send + Sync + Unpin,
1039{
1040    let mut out = Vec::with_capacity(entities.len());
1041    for (index, entity) in entities.iter().enumerate() {
1042        let (query, json) = crate::repository::insert_statement(table, entity)
1043            .map_err(|error| BatchItemError { index, error })?;
1044        let statement = match json {
1045            Some(json) => sqlx::query_as::<_, T>(&query).bind(json),
1046            None => sqlx::query_as::<_, T>(&query),
1047        };
1048        let row = statement.fetch_one(&mut *conn).await.map_err(|e| BatchItemError {
1049            index,
1050            error: crate::repository::explain_unknown_column(e, table),
1051        })?;
1052        out.push(row);
1053    }
1054    Ok(out)
1055}
1056
1057/// Apply each `(before, after)` pair on `conn` (an open transaction) as a changed-columns
1058/// update under `guard`, stopping at the first row that fails or no longer matches.
1059async fn update_each_changed<T>(
1060    conn: &mut sqlx::PgConnection,
1061    table: &str,
1062    guard: &str,
1063    pairs: &[(T, T)],
1064) -> Result<Vec<T>>
1065where
1066    T: for<'r> FromRow<'r, PgRow> + Serialize + Send + Sync + Unpin,
1067{
1068    use crate::repository::UpdateStatement;
1069    let mut out = Vec::with_capacity(pairs.len());
1070    for (index, (before, after)) in pairs.iter().enumerate() {
1071        let item = |error: anyhow::Error| BatchItemError { index, error };
1072        let id = match serde_json::to_value(after).map_err(|e| item(e.into()))? {
1073            serde_json::Value::Object(obj) => match obj.get("id") {
1074                Some(serde_json::Value::String(id)) => id.clone(),
1075                _ => return Err(item(anyhow::anyhow!("entity missing string 'id' field")).into()),
1076            },
1077            _ => return Err(item(anyhow::anyhow!("entity must serialize to a JSON object")).into()),
1078        };
1079        let changed = crate::repository::changed_columns(before, after).map_err(item)?;
1080        let updated = match crate::repository::update_statement(table, after, Some(&changed), guard)
1081            .map_err(item)?
1082        {
1083            UpdateStatement::Unchanged(query) => {
1084                sqlx::query_as::<_, T>(&query).bind(&id).fetch_optional(&mut *conn).await
1085            }
1086            UpdateStatement::Update { query, json } => {
1087                sqlx::query_as::<_, T>(&query)
1088                    .bind(json)
1089                    .bind(&id)
1090                    .fetch_optional(&mut *conn)
1091                    .await
1092            }
1093        }
1094        .map_err(|e| item(e.into()))?;
1095        match updated {
1096            Some(row) => out.push(row),
1097            None => {
1098                return Err(item(anyhow::anyhow!("id '{id}' not found or already deleted")).into())
1099            }
1100        }
1101    }
1102    Ok(out)
1103}
1104
1105/// Serialize an entity and return `(id, json_string, quoted_column_list)` for the
1106/// `jsonb_populate_record` update query — shared by the per-mode `bulk_update`s.
1107fn build_update_parts<T: Serialize>(entity: &T) -> Result<(String, String, String)> {
1108    let json_value = serde_json::to_value(entity)?;
1109    let json_obj = match json_value {
1110        serde_json::Value::Object(obj) => obj,
1111        _ => return Err(anyhow::anyhow!("entity must serialize to a JSON object")),
1112    };
1113    let id = json_obj
1114        .get("id")
1115        .and_then(|v| v.as_str())
1116        .ok_or_else(|| anyhow::anyhow!("entity missing string 'id' field"))?
1117        .to_string();
1118    let column_names = json_obj
1119        .keys()
1120        .filter(|k| *k != "id")
1121        .map(|k| format!("\"{k}\""))
1122        .collect::<Vec<_>>()
1123        .join(", ");
1124    let json_str = serde_json::to_string(&json_obj)?;
1125    Ok((id, json_str, column_names))
1126}
1127
1128// ─── HardDelete mode ─────────────────────────────────────────────────────────
1129
1130impl<T> GenericCrudRepository<T, HardDelete>
1131where
1132    T: for<'r> FromRow<'r, PgRow> + Send + Sync + Unpin + Serialize,
1133{
1134    // ── Unique-field lookups (no soft-delete guard) ───────────────────────────
1135
1136    /// Find an entity by a unique text field (no soft-delete guard).
1137    pub async fn find_by_text_field(&self, field: &str, value: &str) -> Result<Option<T>> {
1138        self.find_by_text_field_with_cond(field, value, "").await
1139    }
1140
1141    /// Check existence by a unique text field.
1142    pub async fn exists_by_text_field(&self, field: &str, value: &str) -> Result<bool> {
1143        self.exists_by_text_field_with_cond(field, value, "").await
1144    }
1145
1146    /// Find an entity by a unique UUID field.
1147    pub async fn find_by_uuid_field(&self, field: &str, value: Uuid) -> Result<Option<T>> {
1148        self.find_by_uuid_field_with_cond(field, value, "").await
1149    }
1150
1151    /// Check existence by a unique UUID field.
1152    pub async fn exists_by_uuid_field(&self, field: &str, value: Uuid) -> Result<bool> {
1153        self.exists_by_uuid_field_with_cond(field, value, "").await
1154    }
1155
1156    // ── Filtered pagination ───────────────────────────────────────────────────
1157
1158    /// Paginate entities with filter and search support.
1159    pub async fn list_paginated_filtered(
1160        &self,
1161        pagination: PaginationParams,
1162        filters: Option<&HashMap<String, String>>,
1163    ) -> Result<PaginatedResult<T>>
1164    where
1165        T: EntityRepoMeta + Send + Sync,
1166    {
1167        let filters_map = filters.cloned().unwrap_or_default();
1168        let column_types = T::column_types();
1169        let search_fields_owned: Vec<&'static str> = T::search_fields().iter().copied().collect();
1170        self.run_filtered_query(pagination, None, &filters_map, &column_types, &search_fields_owned).await
1171    }
1172
1173    /// Group and reduce entities. This mode has no trash to exclude.
1174    pub async fn aggregate_filtered(
1175        &self,
1176        spec: &crate::repository::AggregateSpec,
1177        filters: Option<&HashMap<String, String>>,
1178    ) -> Result<crate::repository::AggregateResult>
1179    where
1180        T: EntityRepoMeta + Send + Sync,
1181    {
1182        let filters_map = filters.cloned().unwrap_or_default();
1183        let column_types = T::column_types();
1184        let search_fields_owned: Vec<&'static str> = T::search_fields().iter().copied().collect();
1185        self.run_aggregate_query(spec, None, &filters_map, &column_types, &search_fields_owned).await
1186    }
1187
1188    /// Find an entity by primary key.
1189    pub async fn find_by_id(&self, id: &str) -> Result<Option<T>> {
1190        self.inner.find_by_id(id).await
1191    }
1192
1193    /// Return all entities.
1194    pub async fn find_all(&self) -> Result<Vec<T>> {
1195        let query = format!("SELECT * FROM {}", self.table_name());
1196        let results = crate::company_scope::fetch_all_scoped(
1197            self.pool(),
1198            sqlx::query_as::<_, T>(&query),
1199        )
1200        .await?;
1201        Ok(results)
1202    }
1203
1204    /// Full update.
1205    pub async fn update(&self, id: &str, entity: &T) -> Result<Option<T>> {
1206        self.inner.update(id, entity).await
1207    }
1208
1209    /// Update a record from `before` (the row as read) to `after`, writing only the columns
1210    /// whose value differs; `Ok(None)` when the row no longer exists. See the soft-delete
1211    /// twin for why a read-modify-write must not write back the columns it did not change.
1212    pub async fn update_changed(&self, id: &str, before: &T, after: &T) -> Result<Option<T>> {
1213        let changed = crate::repository::changed_columns(before, after)?;
1214        self.inner
1215            .update_guarded(id, after, Some(&changed), "")
1216            .await
1217    }
1218
1219    /// Permanently delete an entity by primary key.
1220    pub async fn delete(&self, id: &str) -> Result<bool> {
1221        self.inner.delete(id).await
1222    }
1223
1224    /// Count all entities.
1225    pub async fn count(&self) -> Result<u64> {
1226        let query = format!("SELECT COUNT(*) FROM {}", self.table_name());
1227        let count = crate::company_scope::fetch_one_scalar_scoped(
1228            self.pool(),
1229            sqlx::query_scalar::<_, i64>(&query),
1230        )
1231        .await? as u64;
1232        Ok(count)
1233    }
1234
1235    /// Return `true` if an entity with the given ID exists.
1236    pub async fn exists(&self, id: &str) -> Result<bool> {
1237        let query = format!(
1238            "SELECT 1 FROM {} WHERE id = $1::uuid LIMIT 1",
1239            self.table_name()
1240        );
1241        let result = crate::company_scope::fetch_optional_scalar_scoped(
1242            self.pool(),
1243            sqlx::query_scalar::<_, i32>(&query).bind(id),
1244        )
1245        .await?;
1246        Ok(result.is_some())
1247    }
1248
1249    /// Paginate entities (most-recent-first by ID).
1250    pub async fn list_paginated(&self, pagination: PaginationParams) -> Result<PaginatedResult<T>> {
1251        let offset = pagination.offset();
1252        let limit = pagination.limit();
1253        let query = format!(
1254            "SELECT * FROM {} ORDER BY id DESC LIMIT $1 OFFSET $2",
1255            self.table_name()
1256        );
1257        let data = crate::company_scope::fetch_all_scoped(
1258            self.pool(),
1259            sqlx::query_as::<_, T>(&query).bind(limit as i64).bind(offset as i64),
1260        )
1261        .await?;
1262        let total = self.count().await?;
1263        Ok(PaginatedResult {
1264            data,
1265            pagination: PaginationInfo::new(pagination.page, pagination.per_page, total),
1266        })
1267    }
1268
1269    // ── Atomic batch operations ───────────────────────────────────────────────
1270
1271    /// Hard-delete many rows atomically. Rolls back unless every id matched.
1272    pub async fn bulk_delete(&self, ids: &[String]) -> Result<u64> {
1273        if ids.is_empty() {
1274            return Ok(0);
1275        }
1276        let placeholders = id_in_placeholders(ids.len());
1277        let query = format!(
1278            "DELETE FROM {} WHERE id IN ({placeholders})",
1279            self.table_name()
1280        );
1281        let unit = crate::write_tx::begin_unit(self.pool(), false).await?;
1282        let outcome: Result<u64> = async {
1283            let mut conn = unit.conn().await;
1284            let mut q = sqlx::query(&query);
1285            for id in ids {
1286                q = q.bind(id);
1287            }
1288            let affected = q.execute(&mut *conn).await?.rows_affected();
1289            if affected != ids.len() as u64 {
1290                return Err(anyhow::anyhow!(
1291                    "bulk_delete: {} of {} ids not found; rolled back",
1292                    ids.len() as u64 - affected,
1293                    ids.len()
1294                ));
1295            }
1296            Ok(affected)
1297        }
1298        .await;
1299        unit.finish(outcome).await
1300    }
1301
1302    /// Update many rows atomically. No soft-delete guard in this mode.
1303    pub async fn bulk_update(&self, entities: &[T]) -> Result<Vec<T>> {
1304        bulk_update_rows(self.pool(), self.table_name(), "", entities).await
1305    }
1306
1307    /// Update many rows from `(before, after)` pairs atomically, writing only the columns each
1308    /// update changed; a row deleted since it was read rolls the whole batch back.
1309    pub async fn bulk_update_changed(&self, pairs: &[(T, T)]) -> Result<Vec<T>> {
1310        self.bulk_update_changed_guarded(pairs, "").await
1311    }
1312}
1313
1314/// Qualify a relation target table with the calling entity's schema.
1315///
1316/// `EntityRepoMeta::relations()` emits bare table names (e.g. `countries`), but
1317/// module tables live in the module's own schema (`geo.countries`) and the
1318/// runtime app role's search_path does not include module schemas — an
1319/// unqualified reference fails with "relation does not exist". The calling
1320/// entity's table name is schema-qualified, so its prefix is the schema a
1321/// relation declared inside one module shares with its target. Targets that
1322/// already carry a schema, and callers that do not, pass through unchanged.
1323pub fn qualify_relation_table(caller_table: &str, target_table: &str) -> String {
1324    if target_table.contains('.') || !caller_table.contains('.') {
1325        target_table.to_string()
1326    } else {
1327        let schema = caller_table.split('.').next().unwrap_or(caller_table);
1328        format!("{schema}.{target_table}")
1329    }
1330}
1331
1332/// True when the error is Postgres `undefined_table` (SQLSTATE 42P01).
1333///
1334/// Walks the anyhow chain because the scoped-fetch error is converted with `?`
1335/// on its way up — the sqlx error rides as the (only) source.
1336fn is_undefined_table(err: &anyhow::Error) -> bool {
1337    err.chain()
1338        .filter_map(|cause| cause.downcast_ref::<sqlx::Error>())
1339        .any(|sqlx_err| {
1340            sqlx_err
1341                .as_database_error()
1342                .map(|db| db.code().as_deref() == Some("42P01"))
1343                .unwrap_or(false)
1344        })
1345}
1346
1347/// Fetch rows from an arbitrary table by id list, as JSON — hydrates `?include=`
1348/// relations in the generic CRUD handler without a typed repository for the
1349/// target entity. `table` is a generator-emitted collection name (from
1350/// `EntityRepoMeta::relations()`), NEVER client input, so the interpolation is
1351/// not an injection vector. `caller_table` is the hydrated entity's own
1352/// (schema-qualified) table name, used to qualify a bare `table` — see
1353/// [`qualify_relation_table`]. One batched `WHERE id = ANY(...)` per relation.
1354/// Returns `row_to_json` objects keyed by raw (snake_case) column names.
1355///
1356/// Resolution is two-step for bare targets: the caller's schema first, then —
1357/// only on `undefined_table` — the name as written, which resolves through the
1358/// role's search_path (platform tables such as `users` or `roles` live in
1359/// `public` while their declaring entities live in a module schema). Any other
1360/// error, or a target that fails both spellings, is returned to the caller.
1361pub async fn fetch_by_ids_as_json(
1362    pool: &PgPool,
1363    caller_table: &str,
1364    table: &str,
1365    ids: &[String],
1366) -> Result<Vec<serde_json::Value>> {
1367    if ids.is_empty() {
1368        return Ok(Vec::new());
1369    }
1370    let qualified = qualify_relation_table(caller_table, table);
1371    match fetch_rows_as_json(pool, &qualified, ids).await {
1372        Ok(rows) => Ok(rows),
1373        Err(err) if qualified != table && is_undefined_table(&err) => {
1374            fetch_rows_as_json(pool, table, ids).await
1375        }
1376        Err(err) => Err(err),
1377    }
1378}
1379
1380/// Whether the row of `table` whose `column` holds `value` is visible on the caller's
1381/// scope: one `SELECT EXISTS` run as every read is (on the request's connection when one
1382/// is bound, else under the task's company scope), so a row row-level security hides
1383/// reads as absent. Postgres checks a foreign key without row-level security; this is
1384/// the check that does not.
1385///
1386/// `value` is a JSON scalar, cast to the column's own type by `jsonb_populate_record`, so
1387/// the lookup keeps the column's index whatever its type. `table` and `column` are
1388/// generator-emitted (`PersistentEntity::references`), never client input, and are refused
1389/// unless they are plain identifiers. A bare `table` resolves as
1390/// [`fetch_by_ids_as_json`]'s does: the caller's schema first, then the search path.
1391/// `Ok(None)` when the table does not exist in either spelling (its module is not
1392/// installed, and the database has no key to it either).
1393pub async fn reference_visible(
1394    pool: &PgPool,
1395    caller_table: &str,
1396    table: &str,
1397    column: &str,
1398    value: &serde_json::Value,
1399) -> Result<Option<bool>> {
1400    let plain = |s: &str, dot: bool| {
1401        !s.is_empty() && s.chars().all(|c| c.is_ascii_alphanumeric() || c == '_' || (dot && c == '.'))
1402    };
1403    if !plain(table, true) || !plain(column, false) {
1404        anyhow::bail!("reference target `{table}`.`{column}` is not a plain identifier");
1405    }
1406    let qualified = qualify_relation_table(caller_table, table);
1407    match row_visible(pool, &qualified, column, value).await {
1408        Ok(seen) => Ok(Some(seen)),
1409        Err(err) if is_undefined_table(&err) => {
1410            if qualified == table {
1411                return Ok(None);
1412            }
1413            match row_visible(pool, table, column, value).await {
1414                Ok(seen) => Ok(Some(seen)),
1415                Err(err) if is_undefined_table(&err) => Ok(None),
1416                Err(err) => Err(err),
1417            }
1418        }
1419        Err(err) => Err(err),
1420    }
1421}
1422
1423async fn row_visible(
1424    pool: &PgPool,
1425    table: &str,
1426    column: &str,
1427    value: &serde_json::Value,
1428) -> Result<bool> {
1429    let query = format!(
1430        "SELECT EXISTS (SELECT 1 FROM {table} t \
1431         WHERE t.{column} = (jsonb_populate_record(NULL::{table}, $1::jsonb)).{column})"
1432    );
1433    let probe = serde_json::json!({ column: value });
1434    // A missing table fails this statement and the caller tries another spelling. Inside a write
1435    // transaction that failure would abort it, so the lookup runs as a savepoint of it.
1436    Ok(crate::write_tx::within_savepoint(crate::company_scope::fetch_one_scalar_scoped(
1437        pool,
1438        sqlx::query_scalar::<_, bool>(&query).bind(probe),
1439    ))
1440    .await?)
1441}
1442
1443/// The scoped fetch behind [`fetch_by_ids_as_json`] — one relation, one table
1444/// spelling. Scoped because an `?include=` hydration of a company-fenced
1445/// relation must obey the same fence, so a cross-company related row is never
1446/// leaked through the expansion.
1447async fn fetch_rows_as_json(
1448    pool: &PgPool,
1449    table: &str,
1450    ids: &[String],
1451) -> Result<Vec<serde_json::Value>> {
1452    let query = format!("SELECT row_to_json(t) AS j FROM {table} t WHERE t.id = ANY($1::uuid[])");
1453    let rows: Vec<(serde_json::Value,)> =
1454        crate::company_scope::fetch_all_scoped(pool, sqlx::query_as(&query).bind(ids)).await?;
1455    Ok(rows.into_iter().map(|(j,)| j).collect())
1456}
1457
1458/// Shared transactional bulk-update used by both delete modes. Each entity is
1459/// updated by id inside one transaction; a missing (or, with `active_guard`,
1460/// soft-deleted) row rolls the whole batch back. `active_guard` is an extra SQL
1461/// predicate ANDed into the `WHERE` clause (`""` for none).
1462async fn bulk_update_rows<T>(
1463    pool: &PgPool,
1464    table: &str,
1465    active_guard: &str,
1466    entities: &[T],
1467) -> Result<Vec<T>>
1468where
1469    T: for<'r> FromRow<'r, PgRow> + Send + Sync + Unpin + Serialize,
1470{
1471    if entities.is_empty() {
1472        return Ok(Vec::new());
1473    }
1474    let unit = crate::write_tx::begin_unit(pool, false).await?;
1475    let outcome: Result<Vec<T>> = async {
1476        let mut conn = unit.conn().await;
1477        let mut out = Vec::with_capacity(entities.len());
1478        for entity in entities {
1479            let (id, json_str, column_names) = build_update_parts(entity)?;
1480            let query = format!(
1481                "WITH new_row AS (\
1482                    SELECT (jsonb_populate_record(NULL::{table}, $1::jsonb)).*\
1483                 ) UPDATE {table} AS t \
1484                 SET ({columns}) = (SELECT {columns} FROM new_row) \
1485                 WHERE t.id = $2::uuid{guard} \
1486                 RETURNING t.*",
1487                table = table,
1488                columns = column_names,
1489                guard = active_guard,
1490            );
1491            let updated = sqlx::query_as::<_, T>(&query)
1492                .bind(&json_str)
1493                .bind(&id)
1494                .fetch_optional(&mut *conn)
1495                .await?;
1496            match updated {
1497                Some(e) => out.push(e),
1498                None => {
1499                    return Err(anyhow::anyhow!(
1500                        "bulk_update: id '{id}' not found or already deleted; rolled back"
1501                    ));
1502                }
1503            }
1504        }
1505        Ok(out)
1506    }
1507    .await;
1508    unit.finish(outcome).await
1509}
1510
1511/// camelCase → snake_case (`projectId` → `project_id`), the SQL vocabulary.
1512fn camel_to_snake(s: &str) -> String {
1513    let mut out = String::with_capacity(s.len() + 4);
1514    for ch in s.chars() {
1515        if ch.is_uppercase() {
1516            out.push('_');
1517            out.extend(ch.to_lowercase());
1518        } else {
1519            out.push(ch);
1520        }
1521    }
1522    out
1523}