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()).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()).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, ordered by `id` ascending — the same key order every
702    /// filtered list ends on, so a page never repeats or skips a row between two fetches.
703    pub async fn list_paginated(&self, pagination: PaginationParams) -> Result<PaginatedResult<T>> {
704        let offset = pagination.offset();
705        let limit = pagination.limit();
706        let query = format!(
707            "SELECT * FROM {} WHERE metadata->>'deleted_at' IS NULL \
708             ORDER BY id LIMIT $1 OFFSET $2",
709            self.table_name()
710        );
711        let data = crate::company_scope::fetch_all_scoped(
712            self.pool(),
713            sqlx::query_as::<_, T>(&query).bind(limit as i64).bind(offset as i64),
714        )
715        .await?;
716        let total = self.count_active().await?;
717        Ok(PaginatedResult {
718            data,
719            pagination: PaginationInfo::new(pagination.page, pagination.per_page, total),
720        })
721    }
722
723    // ── Soft-delete helpers ───────────────────────────────────────────────────
724
725    /// Set `metadata.deleted_at` to NOW() (soft delete).
726    pub async fn soft_delete(&self, id: &str) -> Result<bool> {
727        let query = format!(
728            "UPDATE {} SET metadata = jsonb_set(\
729               COALESCE(metadata, '{{}}'), \
730               '{{deleted_at}}', \
731               to_jsonb(NOW())\
732             ) WHERE id = $1::uuid AND (metadata->>'deleted_at') IS NULL",
733            self.table_name()
734        );
735        let result = crate::company_scope::execute_scoped(
736            self.pool(),
737            sqlx::query(&query).bind(id),
738        )
739        .await?;
740        Ok(result.rows_affected() > 0)
741    }
742
743    /// Remove `deleted_at` from metadata, restoring the entity.
744    pub async fn restore(&self, id: &str) -> Result<Option<T>> {
745        let query = format!(
746            "UPDATE {} SET metadata = metadata - 'deleted_at' \
747             WHERE id = $1::uuid AND (metadata->>'deleted_at') IS NOT NULL \
748             RETURNING *",
749            self.table_name()
750        );
751        let result = crate::company_scope::fetch_optional_scoped(
752            self.pool(),
753            sqlx::query_as::<_, T>(&query).bind(id),
754        )
755        .await?;
756        Ok(result)
757    }
758
759    /// Paginate soft-deleted entities (trash view), most recently deleted first, `id` ascending
760    /// among rows deleted at the same instant so the order is total.
761    pub async fn list_deleted(&self, pagination: PaginationParams) -> Result<PaginatedResult<T>> {
762        let offset = pagination.offset();
763        let limit = pagination.limit();
764        let query = format!(
765            "SELECT * FROM {} WHERE (metadata->>'deleted_at') IS NOT NULL \
766             ORDER BY (metadata->>'deleted_at') DESC, id LIMIT $1 OFFSET $2",
767            self.table_name()
768        );
769        let data = crate::company_scope::fetch_all_scoped(
770            self.pool(),
771            sqlx::query_as::<_, T>(&query).bind(limit as i64).bind(offset as i64),
772        )
773        .await?;
774        let count_query = format!(
775            "SELECT COUNT(*) FROM {} WHERE (metadata->>'deleted_at') IS NOT NULL",
776            self.table_name()
777        );
778        let total = crate::company_scope::fetch_one_scalar_scoped(
779            self.pool(),
780            sqlx::query_scalar::<_, i64>(&count_query),
781        )
782        .await? as u64;
783        Ok(PaginatedResult {
784            data,
785            pagination: PaginationInfo::new(pagination.page, pagination.per_page, total),
786        })
787    }
788
789    /// Permanently delete all soft-deleted rows (empty trash).
790    pub async fn empty_trash(&self) -> Result<u64> {
791        let query = format!(
792            "DELETE FROM {} WHERE (metadata->>'deleted_at') IS NOT NULL",
793            self.table_name()
794        );
795        let result = crate::company_scope::execute_scoped(self.pool(), sqlx::query(&query)).await?;
796        Ok(result.rows_affected())
797    }
798
799    /// Find a soft-deleted entity by primary key.
800    pub async fn find_deleted_by_id(&self, id: &str) -> Result<Option<T>> {
801        let query = format!(
802            "SELECT * FROM {} WHERE id = $1::uuid AND (metadata->>'deleted_at') IS NOT NULL",
803            self.table_name()
804        );
805        let result = crate::company_scope::fetch_optional_scoped(
806            self.pool(),
807            sqlx::query_as::<_, T>(&query).bind(id),
808        )
809        .await?;
810        Ok(result)
811    }
812
813    /// Permanently delete a soft-deleted entity by primary key.
814    pub async fn permanent_delete(&self, id: &str) -> Result<bool> {
815        let query = format!(
816            "DELETE FROM {} WHERE id = $1::uuid AND (metadata->>'deleted_at') IS NOT NULL",
817            self.table_name()
818        );
819        let result = crate::company_scope::execute_scoped(
820            self.pool(),
821            sqlx::query(&query).bind(id),
822        )
823        .await?;
824        Ok(result.rows_affected() > 0)
825    }
826
827    /// Count active (non-deleted) entities.
828    pub async fn count_active(&self) -> Result<u64> {
829        let query = format!(
830            "SELECT COUNT(*) FROM {} WHERE (metadata->>'deleted_at') IS NULL",
831            self.table_name()
832        );
833        let count = crate::company_scope::fetch_one_scalar_scoped(
834            self.pool(),
835            sqlx::query_scalar::<_, i64>(&query),
836        )
837        .await? as u64;
838        Ok(count)
839    }
840
841    /// Count soft-deleted entities.
842    pub async fn count_deleted(&self) -> Result<u64> {
843        let query = format!(
844            "SELECT COUNT(*) FROM {} WHERE (metadata->>'deleted_at') IS NOT NULL",
845            self.table_name()
846        );
847        let count = crate::company_scope::fetch_one_scalar_scoped(
848            self.pool(),
849            sqlx::query_scalar::<_, i64>(&query),
850        )
851        .await? as u64;
852        Ok(count)
853    }
854
855    // ── Atomic batch operations ───────────────────────────────────────────────
856    //
857    // Each runs inside a single transaction. For id-list operations the affected
858    // row count must equal the number of ids requested, otherwise the whole batch
859    // is rolled back — all-or-nothing semantics where a missing / already-in-the-
860    // target-state id (or a duplicate id) fails the entire request.
861
862    /// Soft-delete many active rows atomically.
863    pub async fn bulk_soft_delete(&self, ids: &[String]) -> Result<u64> {
864        if ids.is_empty() {
865            return Ok(0);
866        }
867        let placeholders = id_in_placeholders(ids.len());
868        let query = format!(
869            "UPDATE {} SET metadata = jsonb_set(\
870               COALESCE(metadata, '{{}}'), '{{deleted_at}}', to_jsonb(NOW())\
871             ) WHERE id IN ({placeholders}) AND (metadata->>'deleted_at') IS NULL",
872            self.table_name()
873        );
874        let unit = crate::write_tx::begin_unit(self.pool()).await?;
875        let outcome: Result<u64> = async {
876            let mut conn = unit.conn().await;
877            let mut q = sqlx::query(&query);
878            for id in ids {
879                q = q.bind(id);
880            }
881            let affected = q.execute(&mut *conn).await?.rows_affected();
882            if affected != ids.len() as u64 {
883                // Finishing with an error rolls the unit back.
884                return Err(anyhow::anyhow!(
885                    "bulk_soft_delete: {} of {} ids were not active/deletable; rolled back",
886                    ids.len() as u64 - affected,
887                    ids.len()
888                ));
889            }
890            Ok(affected)
891        }
892        .await;
893        unit.finish(outcome).await
894    }
895
896    /// Restore many soft-deleted rows atomically, returning the restored rows.
897    pub async fn bulk_restore(&self, ids: &[String]) -> Result<Vec<T>> {
898        if ids.is_empty() {
899            return Ok(Vec::new());
900        }
901        let placeholders = id_in_placeholders(ids.len());
902        let query = format!(
903            "UPDATE {} SET metadata = metadata - 'deleted_at' \
904             WHERE id IN ({placeholders}) AND (metadata->>'deleted_at') IS NOT NULL \
905             RETURNING *",
906            self.table_name()
907        );
908        let unit = crate::write_tx::begin_unit(self.pool()).await?;
909        let outcome: Result<Vec<T>> = async {
910            let mut conn = unit.conn().await;
911            let mut q = sqlx::query_as::<_, T>(&query);
912            for id in ids {
913                q = q.bind(id);
914            }
915            let rows = q.fetch_all(&mut *conn).await?;
916            if rows.len() != ids.len() {
917                return Err(anyhow::anyhow!(
918                    "bulk_restore: {} of {} ids were not in trash; rolled back",
919                    ids.len() - rows.len(),
920                    ids.len()
921                ));
922            }
923            Ok(rows)
924        }
925        .await;
926        unit.finish(outcome).await
927    }
928
929    /// Permanently delete many soft-deleted rows atomically.
930    pub async fn bulk_permanent_delete(&self, ids: &[String]) -> Result<u64> {
931        if ids.is_empty() {
932            return Ok(0);
933        }
934        let placeholders = id_in_placeholders(ids.len());
935        let query = format!(
936            "DELETE FROM {} WHERE id IN ({placeholders}) AND (metadata->>'deleted_at') IS NOT NULL",
937            self.table_name()
938        );
939        let unit = crate::write_tx::begin_unit(self.pool()).await?;
940        let outcome: Result<u64> = async {
941            let mut conn = unit.conn().await;
942            let mut q = sqlx::query(&query);
943            for id in ids {
944                q = q.bind(id);
945            }
946            let affected = q.execute(&mut *conn).await?.rows_affected();
947            if affected != ids.len() as u64 {
948                return Err(anyhow::anyhow!(
949                    "bulk_permanent_delete: {} of {} ids were not in trash; rolled back",
950                    ids.len() as u64 - affected,
951                    ids.len()
952                ));
953            }
954            Ok(affected)
955        }
956        .await;
957        unit.finish(outcome).await
958    }
959
960    /// Restore every soft-deleted row, returning the restored rows. A single
961    /// `UPDATE ... RETURNING` is atomic, and the returned rows let the service
962    /// layer emit a `Restored` event per entity.
963    pub async fn restore_all(&self) -> Result<Vec<T>> {
964        let query = format!(
965            "UPDATE {} SET metadata = metadata - 'deleted_at' \
966             WHERE (metadata->>'deleted_at') IS NOT NULL \
967             RETURNING *",
968            self.table_name()
969        );
970        let rows = crate::company_scope::fetch_all_scoped(
971            self.pool(),
972            sqlx::query_as::<_, T>(&query),
973        )
974        .await?;
975        Ok(rows)
976    }
977
978    /// Update many active rows atomically. Every entity must reference an
979    /// existing active (non-soft-deleted) row or the whole batch is rolled back.
980    pub async fn bulk_update(&self, entities: &[T]) -> Result<Vec<T>>
981    where
982        T: Serialize + Send + Sync,
983    {
984        bulk_update_rows(
985            self.pool(),
986            self.table_name(),
987            ACTIVE_ROW_GUARD,
988            entities,
989        )
990        .await
991    }
992
993    /// Update many active rows from `(before, after)` pairs atomically, writing only the
994    /// columns each update changed; a row missing or soft-deleted since it was read rolls the
995    /// whole batch back.
996    pub async fn bulk_update_changed(&self, pairs: &[(T, T)]) -> Result<Vec<T>>
997    where
998        T: Serialize + Send + Sync,
999    {
1000        self.bulk_update_changed_guarded(pairs, ACTIVE_ROW_GUARD).await
1001    }
1002}
1003
1004/// Build `"$1::uuid, $2::uuid, …"` for an `id IN (…)` clause of `n` bound ids.
1005fn id_in_placeholders(n: usize) -> String {
1006    (1..=n)
1007        .map(|i| format!("${i}::uuid"))
1008        .collect::<Vec<_>>()
1009        .join(", ")
1010}
1011
1012/// One item of a batch write failed, so the whole batch was rolled back.
1013///
1014/// `index` is the item's position in the batch as the caller passed it; `error` is why it
1015/// failed (a database refusal keeps its `sqlx::Error` inside, so a caller can still read the
1016/// constraint it broke).
1017#[derive(Debug)]
1018pub struct BatchItemError {
1019    /// Position of the failed item in the batch.
1020    pub index: usize,
1021    /// Why it failed.
1022    pub error: anyhow::Error,
1023}
1024
1025impl std::fmt::Display for BatchItemError {
1026    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
1027        write!(f, "item {} of the batch: {}; rolled back", self.index, self.error)
1028    }
1029}
1030
1031impl std::error::Error for BatchItemError {
1032    fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
1033        Some(self.error.as_ref())
1034    }
1035}
1036
1037/// Insert each entity on `conn` (an open transaction), stopping at the first failure.
1038async fn insert_each<T>(conn: &mut sqlx::PgConnection, table: &str, entities: &[T]) -> Result<Vec<T>>
1039where
1040    T: for<'r> FromRow<'r, PgRow> + Serialize + Send + Sync + Unpin,
1041{
1042    let mut out = Vec::with_capacity(entities.len());
1043    for (index, entity) in entities.iter().enumerate() {
1044        let (query, json) = crate::repository::insert_statement(table, entity)
1045            .map_err(|error| BatchItemError { index, error })?;
1046        let statement = match json {
1047            Some(json) => sqlx::query_as::<_, T>(&query).bind(json),
1048            None => sqlx::query_as::<_, T>(&query),
1049        };
1050        let row = statement.fetch_one(&mut *conn).await.map_err(|e| BatchItemError {
1051            index,
1052            error: crate::repository::explain_unknown_column(e, table),
1053        })?;
1054        out.push(row);
1055    }
1056    Ok(out)
1057}
1058
1059/// Apply each `(before, after)` pair on `conn` (an open transaction) as a changed-columns
1060/// update under `guard`, stopping at the first row that fails or no longer matches.
1061async fn update_each_changed<T>(
1062    conn: &mut sqlx::PgConnection,
1063    table: &str,
1064    guard: &str,
1065    pairs: &[(T, T)],
1066) -> Result<Vec<T>>
1067where
1068    T: for<'r> FromRow<'r, PgRow> + Serialize + Send + Sync + Unpin,
1069{
1070    use crate::repository::UpdateStatement;
1071    let mut out = Vec::with_capacity(pairs.len());
1072    for (index, (before, after)) in pairs.iter().enumerate() {
1073        let item = |error: anyhow::Error| BatchItemError { index, error };
1074        let id = match serde_json::to_value(after).map_err(|e| item(e.into()))? {
1075            serde_json::Value::Object(obj) => match obj.get("id") {
1076                Some(serde_json::Value::String(id)) => id.clone(),
1077                _ => return Err(item(anyhow::anyhow!("entity missing string 'id' field")).into()),
1078            },
1079            _ => return Err(item(anyhow::anyhow!("entity must serialize to a JSON object")).into()),
1080        };
1081        let changed = crate::repository::changed_columns(before, after).map_err(item)?;
1082        let updated = match crate::repository::update_statement(table, after, Some(&changed), guard)
1083            .map_err(item)?
1084        {
1085            UpdateStatement::Unchanged(query) => {
1086                sqlx::query_as::<_, T>(&query).bind(&id).fetch_optional(&mut *conn).await
1087            }
1088            UpdateStatement::Update { query, json } => {
1089                sqlx::query_as::<_, T>(&query)
1090                    .bind(json)
1091                    .bind(&id)
1092                    .fetch_optional(&mut *conn)
1093                    .await
1094            }
1095        }
1096        .map_err(|e| item(e.into()))?;
1097        match updated {
1098            Some(row) => out.push(row),
1099            None => {
1100                return Err(item(anyhow::anyhow!("id '{id}' not found or already deleted")).into())
1101            }
1102        }
1103    }
1104    Ok(out)
1105}
1106
1107/// Serialize an entity and return `(id, json_string, quoted_column_list)` for the
1108/// `jsonb_populate_record` update query — shared by the per-mode `bulk_update`s.
1109fn build_update_parts<T: Serialize>(entity: &T) -> Result<(String, String, String)> {
1110    let json_value = serde_json::to_value(entity)?;
1111    let json_obj = match json_value {
1112        serde_json::Value::Object(obj) => obj,
1113        _ => return Err(anyhow::anyhow!("entity must serialize to a JSON object")),
1114    };
1115    let id = json_obj
1116        .get("id")
1117        .and_then(|v| v.as_str())
1118        .ok_or_else(|| anyhow::anyhow!("entity missing string 'id' field"))?
1119        .to_string();
1120    let column_names = json_obj
1121        .keys()
1122        .filter(|k| *k != "id")
1123        .map(|k| format!("\"{k}\""))
1124        .collect::<Vec<_>>()
1125        .join(", ");
1126    let json_str = serde_json::to_string(&json_obj)?;
1127    Ok((id, json_str, column_names))
1128}
1129
1130// ─── HardDelete mode ─────────────────────────────────────────────────────────
1131
1132impl<T> GenericCrudRepository<T, HardDelete>
1133where
1134    T: for<'r> FromRow<'r, PgRow> + Send + Sync + Unpin + Serialize,
1135{
1136    // ── Unique-field lookups (no soft-delete guard) ───────────────────────────
1137
1138    /// Find an entity by a unique text field (no soft-delete guard).
1139    pub async fn find_by_text_field(&self, field: &str, value: &str) -> Result<Option<T>> {
1140        self.find_by_text_field_with_cond(field, value, "").await
1141    }
1142
1143    /// Check existence by a unique text field.
1144    pub async fn exists_by_text_field(&self, field: &str, value: &str) -> Result<bool> {
1145        self.exists_by_text_field_with_cond(field, value, "").await
1146    }
1147
1148    /// Find an entity by a unique UUID field.
1149    pub async fn find_by_uuid_field(&self, field: &str, value: Uuid) -> Result<Option<T>> {
1150        self.find_by_uuid_field_with_cond(field, value, "").await
1151    }
1152
1153    /// Check existence by a unique UUID field.
1154    pub async fn exists_by_uuid_field(&self, field: &str, value: Uuid) -> Result<bool> {
1155        self.exists_by_uuid_field_with_cond(field, value, "").await
1156    }
1157
1158    // ── Filtered pagination ───────────────────────────────────────────────────
1159
1160    /// Paginate entities with filter and search support.
1161    pub async fn list_paginated_filtered(
1162        &self,
1163        pagination: PaginationParams,
1164        filters: Option<&HashMap<String, String>>,
1165    ) -> Result<PaginatedResult<T>>
1166    where
1167        T: EntityRepoMeta + Send + Sync,
1168    {
1169        let filters_map = filters.cloned().unwrap_or_default();
1170        let column_types = T::column_types();
1171        let search_fields_owned: Vec<&'static str> = T::search_fields().iter().copied().collect();
1172        self.run_filtered_query(pagination, None, &filters_map, &column_types, &search_fields_owned).await
1173    }
1174
1175    /// Group and reduce entities. This mode has no trash to exclude.
1176    pub async fn aggregate_filtered(
1177        &self,
1178        spec: &crate::repository::AggregateSpec,
1179        filters: Option<&HashMap<String, String>>,
1180    ) -> Result<crate::repository::AggregateResult>
1181    where
1182        T: EntityRepoMeta + Send + Sync,
1183    {
1184        let filters_map = filters.cloned().unwrap_or_default();
1185        let column_types = T::column_types();
1186        let search_fields_owned: Vec<&'static str> = T::search_fields().iter().copied().collect();
1187        self.run_aggregate_query(spec, None, &filters_map, &column_types, &search_fields_owned).await
1188    }
1189
1190    /// Find an entity by primary key.
1191    pub async fn find_by_id(&self, id: &str) -> Result<Option<T>> {
1192        self.inner.find_by_id(id).await
1193    }
1194
1195    /// Return all entities.
1196    pub async fn find_all(&self) -> Result<Vec<T>> {
1197        let query = format!("SELECT * FROM {}", self.table_name());
1198        let results = crate::company_scope::fetch_all_scoped(
1199            self.pool(),
1200            sqlx::query_as::<_, T>(&query),
1201        )
1202        .await?;
1203        Ok(results)
1204    }
1205
1206    /// Full update.
1207    pub async fn update(&self, id: &str, entity: &T) -> Result<Option<T>> {
1208        self.inner.update(id, entity).await
1209    }
1210
1211    /// Update a record from `before` (the row as read) to `after`, writing only the columns
1212    /// whose value differs; `Ok(None)` when the row no longer exists. See the soft-delete
1213    /// twin for why a read-modify-write must not write back the columns it did not change.
1214    pub async fn update_changed(&self, id: &str, before: &T, after: &T) -> Result<Option<T>> {
1215        let changed = crate::repository::changed_columns(before, after)?;
1216        self.inner
1217            .update_guarded(id, after, Some(&changed), "")
1218            .await
1219    }
1220
1221    /// Permanently delete an entity by primary key.
1222    pub async fn delete(&self, id: &str) -> Result<bool> {
1223        self.inner.delete(id).await
1224    }
1225
1226    /// Count all entities.
1227    pub async fn count(&self) -> Result<u64> {
1228        let query = format!("SELECT COUNT(*) FROM {}", self.table_name());
1229        let count = crate::company_scope::fetch_one_scalar_scoped(
1230            self.pool(),
1231            sqlx::query_scalar::<_, i64>(&query),
1232        )
1233        .await? as u64;
1234        Ok(count)
1235    }
1236
1237    /// Return `true` if an entity with the given ID exists.
1238    pub async fn exists(&self, id: &str) -> Result<bool> {
1239        let query = format!(
1240            "SELECT 1 FROM {} WHERE id = $1::uuid LIMIT 1",
1241            self.table_name()
1242        );
1243        let result = crate::company_scope::fetch_optional_scalar_scoped(
1244            self.pool(),
1245            sqlx::query_scalar::<_, i32>(&query).bind(id),
1246        )
1247        .await?;
1248        Ok(result.is_some())
1249    }
1250
1251    /// Paginate entities, ordered by `id` ascending — the same key order every filtered list
1252    /// ends on, so a page never repeats or skips a row between two fetches.
1253    pub async fn list_paginated(&self, pagination: PaginationParams) -> Result<PaginatedResult<T>> {
1254        let offset = pagination.offset();
1255        let limit = pagination.limit();
1256        let query = format!(
1257            "SELECT * FROM {} ORDER BY id LIMIT $1 OFFSET $2",
1258            self.table_name()
1259        );
1260        let data = crate::company_scope::fetch_all_scoped(
1261            self.pool(),
1262            sqlx::query_as::<_, T>(&query).bind(limit as i64).bind(offset as i64),
1263        )
1264        .await?;
1265        let total = self.count().await?;
1266        Ok(PaginatedResult {
1267            data,
1268            pagination: PaginationInfo::new(pagination.page, pagination.per_page, total),
1269        })
1270    }
1271
1272    // ── Atomic batch operations ───────────────────────────────────────────────
1273
1274    /// Hard-delete many rows atomically. Rolls back unless every id matched.
1275    pub async fn bulk_delete(&self, ids: &[String]) -> Result<u64> {
1276        if ids.is_empty() {
1277            return Ok(0);
1278        }
1279        let placeholders = id_in_placeholders(ids.len());
1280        let query = format!(
1281            "DELETE FROM {} WHERE id IN ({placeholders})",
1282            self.table_name()
1283        );
1284        let unit = crate::write_tx::begin_unit(self.pool()).await?;
1285        let outcome: Result<u64> = async {
1286            let mut conn = unit.conn().await;
1287            let mut q = sqlx::query(&query);
1288            for id in ids {
1289                q = q.bind(id);
1290            }
1291            let affected = q.execute(&mut *conn).await?.rows_affected();
1292            if affected != ids.len() as u64 {
1293                return Err(anyhow::anyhow!(
1294                    "bulk_delete: {} of {} ids not found; rolled back",
1295                    ids.len() as u64 - affected,
1296                    ids.len()
1297                ));
1298            }
1299            Ok(affected)
1300        }
1301        .await;
1302        unit.finish(outcome).await
1303    }
1304
1305    /// Update many rows atomically. No soft-delete guard in this mode.
1306    pub async fn bulk_update(&self, entities: &[T]) -> Result<Vec<T>> {
1307        bulk_update_rows(self.pool(), self.table_name(), "", entities).await
1308    }
1309
1310    /// Update many rows from `(before, after)` pairs atomically, writing only the columns each
1311    /// update changed; a row deleted since it was read rolls the whole batch back.
1312    pub async fn bulk_update_changed(&self, pairs: &[(T, T)]) -> Result<Vec<T>> {
1313        self.bulk_update_changed_guarded(pairs, "").await
1314    }
1315}
1316
1317/// Qualify a relation target table with the calling entity's schema.
1318///
1319/// `EntityRepoMeta::relations()` emits bare table names (e.g. `countries`), but
1320/// module tables live in the module's own schema (`geo.countries`) and the
1321/// runtime app role's search_path does not include module schemas — an
1322/// unqualified reference fails with "relation does not exist". The calling
1323/// entity's table name is schema-qualified, so its prefix is the schema a
1324/// relation declared inside one module shares with its target. Targets that
1325/// already carry a schema, and callers that do not, pass through unchanged.
1326pub fn qualify_relation_table(caller_table: &str, target_table: &str) -> String {
1327    if target_table.contains('.') || !caller_table.contains('.') {
1328        target_table.to_string()
1329    } else {
1330        let schema = caller_table.split('.').next().unwrap_or(caller_table);
1331        format!("{schema}.{target_table}")
1332    }
1333}
1334
1335/// True when the error is Postgres `undefined_table` (SQLSTATE 42P01).
1336///
1337/// Walks the anyhow chain because the scoped-fetch error is converted with `?`
1338/// on its way up — the sqlx error rides as the (only) source.
1339fn is_undefined_table(err: &anyhow::Error) -> bool {
1340    err.chain()
1341        .filter_map(|cause| cause.downcast_ref::<sqlx::Error>())
1342        .any(|sqlx_err| {
1343            sqlx_err
1344                .as_database_error()
1345                .map(|db| db.code().as_deref() == Some("42P01"))
1346                .unwrap_or(false)
1347        })
1348}
1349
1350/// Fetch rows from an arbitrary table by id list, as JSON — hydrates `?include=`
1351/// relations in the generic CRUD handler without a typed repository for the
1352/// target entity. `table` is a generator-emitted collection name (from
1353/// `EntityRepoMeta::relations()`), NEVER client input, so the interpolation is
1354/// not an injection vector. `caller_table` is the hydrated entity's own
1355/// (schema-qualified) table name, used to qualify a bare `table` — see
1356/// [`qualify_relation_table`]. One batched `WHERE id = ANY(...)` per relation.
1357/// Returns `row_to_json` objects keyed by raw (snake_case) column names.
1358///
1359/// Resolution is two-step for bare targets: the caller's schema first, then —
1360/// only on `undefined_table` — the name as written, which resolves through the
1361/// role's search_path (platform tables such as `users` or `roles` live in
1362/// `public` while their declaring entities live in a module schema). Any other
1363/// error, or a target that fails both spellings, is returned to the caller.
1364pub async fn fetch_by_ids_as_json(
1365    pool: &PgPool,
1366    caller_table: &str,
1367    table: &str,
1368    ids: &[String],
1369) -> Result<Vec<serde_json::Value>> {
1370    if ids.is_empty() {
1371        return Ok(Vec::new());
1372    }
1373    let qualified = qualify_relation_table(caller_table, table);
1374    match fetch_rows_as_json(pool, &qualified, ids).await {
1375        Ok(rows) => Ok(rows),
1376        Err(err) if qualified != table && is_undefined_table(&err) => {
1377            fetch_rows_as_json(pool, table, ids).await
1378        }
1379        Err(err) => Err(err),
1380    }
1381}
1382
1383/// Whether the row of `table` whose `column` holds `value` is visible on the caller's
1384/// scope: one `SELECT EXISTS` run as every read is (on the request's connection when one
1385/// is bound, else under the task's company scope), so a row row-level security hides
1386/// reads as absent. Postgres checks a foreign key without row-level security; this is
1387/// the check that does not.
1388///
1389/// `value` is a JSON scalar, cast to the column's own type by `jsonb_populate_record`, so
1390/// the lookup keeps the column's index whatever its type. `table` and `column` are
1391/// generator-emitted (`PersistentEntity::references`), never client input, and are refused
1392/// unless they are plain identifiers. A bare `table` resolves as
1393/// [`fetch_by_ids_as_json`]'s does: the caller's schema first, then the search path.
1394/// `Ok(None)` when the table does not exist in either spelling (its module is not
1395/// installed, and the database has no key to it either).
1396pub async fn reference_visible(
1397    pool: &PgPool,
1398    caller_table: &str,
1399    table: &str,
1400    column: &str,
1401    value: &serde_json::Value,
1402) -> Result<Option<bool>> {
1403    let plain = |s: &str, dot: bool| {
1404        !s.is_empty() && s.chars().all(|c| c.is_ascii_alphanumeric() || c == '_' || (dot && c == '.'))
1405    };
1406    if !plain(table, true) || !plain(column, false) {
1407        anyhow::bail!("reference target `{table}`.`{column}` is not a plain identifier");
1408    }
1409    let qualified = qualify_relation_table(caller_table, table);
1410    match row_visible(pool, &qualified, column, value).await {
1411        Ok(seen) => Ok(Some(seen)),
1412        Err(err) if is_undefined_table(&err) => {
1413            if qualified == table {
1414                return Ok(None);
1415            }
1416            match row_visible(pool, table, column, value).await {
1417                Ok(seen) => Ok(Some(seen)),
1418                Err(err) if is_undefined_table(&err) => Ok(None),
1419                Err(err) => Err(err),
1420            }
1421        }
1422        Err(err) => Err(err),
1423    }
1424}
1425
1426async fn row_visible(
1427    pool: &PgPool,
1428    table: &str,
1429    column: &str,
1430    value: &serde_json::Value,
1431) -> Result<bool> {
1432    let query = format!(
1433        "SELECT EXISTS (SELECT 1 FROM {table} t \
1434         WHERE t.{column} = (jsonb_populate_record(NULL::{table}, $1::jsonb)).{column})"
1435    );
1436    let probe = serde_json::json!({ column: value });
1437    // A missing table fails this statement and the caller tries another spelling. Inside a write
1438    // transaction that failure would abort it, so the lookup runs as a savepoint of it.
1439    Ok(crate::write_tx::within_savepoint(crate::company_scope::fetch_one_scalar_scoped(
1440        pool,
1441        sqlx::query_scalar::<_, bool>(&query).bind(probe),
1442    ))
1443    .await?)
1444}
1445
1446/// The scoped fetch behind [`fetch_by_ids_as_json`] — one relation, one table
1447/// spelling. Scoped because an `?include=` hydration of a company-fenced
1448/// relation must obey the same fence, so a cross-company related row is never
1449/// leaked through the expansion.
1450async fn fetch_rows_as_json(
1451    pool: &PgPool,
1452    table: &str,
1453    ids: &[String],
1454) -> Result<Vec<serde_json::Value>> {
1455    let query = format!("SELECT row_to_json(t) AS j FROM {table} t WHERE t.id = ANY($1::uuid[])");
1456    let rows: Vec<(serde_json::Value,)> =
1457        crate::company_scope::fetch_all_scoped(pool, sqlx::query_as(&query).bind(ids)).await?;
1458    Ok(rows.into_iter().map(|(j,)| j).collect())
1459}
1460
1461/// Shared transactional bulk-update used by both delete modes. Each entity is
1462/// updated by id inside one transaction; a missing (or, with `active_guard`,
1463/// soft-deleted) row rolls the whole batch back. `active_guard` is an extra SQL
1464/// predicate ANDed into the `WHERE` clause (`""` for none).
1465async fn bulk_update_rows<T>(
1466    pool: &PgPool,
1467    table: &str,
1468    active_guard: &str,
1469    entities: &[T],
1470) -> Result<Vec<T>>
1471where
1472    T: for<'r> FromRow<'r, PgRow> + Send + Sync + Unpin + Serialize,
1473{
1474    if entities.is_empty() {
1475        return Ok(Vec::new());
1476    }
1477    let unit = crate::write_tx::begin_unit(pool).await?;
1478    let outcome: Result<Vec<T>> = async {
1479        let mut conn = unit.conn().await;
1480        let mut out = Vec::with_capacity(entities.len());
1481        for entity in entities {
1482            let (id, json_str, column_names) = build_update_parts(entity)?;
1483            let query = format!(
1484                "WITH new_row AS (\
1485                    SELECT (jsonb_populate_record(NULL::{table}, $1::jsonb)).*\
1486                 ) UPDATE {table} AS t \
1487                 SET ({columns}) = (SELECT {columns} FROM new_row) \
1488                 WHERE t.id = $2::uuid{guard} \
1489                 RETURNING t.*",
1490                table = table,
1491                columns = column_names,
1492                guard = active_guard,
1493            );
1494            let updated = sqlx::query_as::<_, T>(&query)
1495                .bind(&json_str)
1496                .bind(&id)
1497                .fetch_optional(&mut *conn)
1498                .await?;
1499            match updated {
1500                Some(e) => out.push(e),
1501                None => {
1502                    return Err(anyhow::anyhow!(
1503                        "bulk_update: id '{id}' not found or already deleted; rolled back"
1504                    ));
1505                }
1506            }
1507        }
1508        Ok(out)
1509    }
1510    .await;
1511    unit.finish(outcome).await
1512}
1513
1514/// camelCase → snake_case (`projectId` → `project_id`), the SQL vocabulary.
1515fn camel_to_snake(s: &str) -> String {
1516    let mut out = String::with_capacity(s.len() + 4);
1517    for ch in s.chars() {
1518        if ch.is_uppercase() {
1519            out.push('_');
1520            out.extend(ch.to_lowercase());
1521        } else {
1522            out.push(ch);
1523        }
1524    }
1525    out
1526}