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        if let Some(conn) = crate::company_scope::current_request_conn() {
280            let mut held = conn.lock().await;
281            let mut tx = sqlx::Connection::begin(&mut **held).await?;
282            let rows = insert_each(&mut tx, table, entities).await?;
283            tx.commit().await?;
284            return Ok(rows);
285        }
286        let mut tx = self.pool().begin().await?;
287        crate::company_scope::bind_current_company(&mut tx).await?;
288        let rows = insert_each(&mut tx, table, entities).await?;
289        tx.commit().await?;
290        Ok(rows)
291    }
292
293    /// Update many rows from `(before, after)` pairs inside one transaction, writing for each
294    /// only the columns that differ — the batch twin of a single changed-columns update — and
295    /// only while `guard` holds (see [`crate::repository::PostgresRepository::update_guarded`]).
296    ///
297    /// A row that no longer matches rolls the whole batch back with a [`BatchItemError`]
298    /// naming its position.
299    async fn bulk_update_changed_guarded(&self, pairs: &[(T, T)], guard: &str) -> Result<Vec<T>>
300    where
301        T: Serialize + Send + Sync,
302    {
303        if pairs.is_empty() {
304            return Ok(Vec::new());
305        }
306        let table = self.table_name();
307        if let Some(conn) = crate::company_scope::current_request_conn() {
308            let mut held = conn.lock().await;
309            let mut tx = sqlx::Connection::begin(&mut **held).await?;
310            let rows = update_each_changed(&mut tx, table, guard, pairs).await?;
311            tx.commit().await?;
312            return Ok(rows);
313        }
314        let mut tx = self.pool().begin().await?;
315        crate::company_scope::bind_current_company(&mut tx).await?;
316        let rows = update_each_changed(&mut tx, table, guard, pairs).await?;
317        tx.commit().await?;
318        Ok(rows)
319    }
320
321    // ── Generic unique-field lookups ──────────────────────────────────────────
322    //
323    // These are shared across both delete modes. The SQL is the same for both;
324    // the soft-delete guard is added separately by the per-mode wrappers below.
325
326    /// Internal: query by a text field with a caller-supplied extra condition.
327    async fn find_by_text_field_with_cond(
328        &self,
329        field: &str,
330        value: &str,
331        extra: &str,
332    ) -> Result<Option<T>> {
333        let query = format!(
334            "SELECT * FROM {} WHERE {} = $1{}",
335            self.table_name(), field, extra
336        );
337        let result = crate::company_scope::fetch_optional_scoped(
338            self.pool(),
339            sqlx::query_as::<_, T>(&query).bind(value),
340        )
341        .await?;
342        Ok(result)
343    }
344
345    /// Internal: existence check by a text field with a caller-supplied extra condition.
346    async fn exists_by_text_field_with_cond(
347        &self,
348        field: &str,
349        value: &str,
350        extra: &str,
351    ) -> Result<bool> {
352        let query = format!(
353            "SELECT 1 FROM {} WHERE {} = $1{} LIMIT 1",
354            self.table_name(), field, extra
355        );
356        let result = crate::company_scope::fetch_optional_scalar_scoped(
357            self.pool(),
358            sqlx::query_scalar::<_, i32>(&query).bind(value),
359        )
360        .await?;
361        Ok(result.is_some())
362    }
363
364    /// Internal: query by a UUID field with a caller-supplied extra condition.
365    async fn find_by_uuid_field_with_cond(
366        &self,
367        field: &str,
368        value: Uuid,
369        extra: &str,
370    ) -> Result<Option<T>> {
371        let query = format!(
372            "SELECT * FROM {} WHERE {} = $1{}",
373            self.table_name(), field, extra
374        );
375        let result = crate::company_scope::fetch_optional_scoped(
376            self.pool(),
377            sqlx::query_as::<_, T>(&query).bind(value),
378        )
379        .await?;
380        Ok(result)
381    }
382
383    /// Internal: existence check by a UUID field with a caller-supplied extra condition.
384    async fn exists_by_uuid_field_with_cond(
385        &self,
386        field: &str,
387        value: Uuid,
388        extra: &str,
389    ) -> Result<bool> {
390        let query = format!(
391            "SELECT 1 FROM {} WHERE {} = $1{} LIMIT 1",
392            self.table_name(), field, extra
393        );
394        let result = crate::company_scope::fetch_optional_scalar_scoped(
395            self.pool(),
396            sqlx::query_scalar::<_, i32>(&query).bind(value),
397        )
398        .await?;
399        Ok(result.is_some())
400    }
401
402    /// Execute a filtered / paginated query against this entity's table.
403    ///
404    /// `base_condition` — when `Some`, raw SQL ANDed as its own parenthesised group ahead of
405    /// the caller-supplied filters, which can therefore never widen past it (not even with
406    /// an `[or]` filter). It must be code-authored; the filters map cannot carry one, since
407    /// the parser ignores every `__`-prefixed key.
408    ///
409    /// NOTE: prefer the `*_scoped` entry points for anything a client can reach —
410    /// they compose the company fence into this condition. This raw form applies no
411    /// fence and is for internal/job callers that have already established scope.
412    pub async fn run_filtered_query(
413        &self,
414        pagination: PaginationParams,
415        base_condition: Option<&str>,
416        filters: &HashMap<String, String>,
417        column_types: &HashMap<String, String>,
418        search_fields: &[&str],
419    ) -> Result<PaginatedResult<T>>
420    where
421        T: Send + Sync,
422    {
423        self.inner
424            .list_with_filters_and_base(
425                pagination,
426                base_condition,
427                filters,
428                column_types,
429                search_fields,
430            )
431            .await
432    }
433
434    /// The aggregate twin of [`Self::run_filtered_query`].
435    ///
436    /// Takes the same `base_condition` so an aggregate is computed over exactly
437    /// the row set the matching list would return — a total that counted
438    /// soft-deleted rows, or another tenant's, would be wrong in a way no
439    /// caller could see.
440    pub async fn run_aggregate_query(
441        &self,
442        spec: &crate::repository::AggregateSpec,
443        base_condition: Option<&str>,
444        filters: &HashMap<String, String>,
445        column_types: &HashMap<String, String>,
446        search_fields: &[&str],
447    ) -> Result<crate::repository::AggregateResult>
448    where
449        T: crate::EntityRepoMeta + Send + Sync,
450    {
451        // Resolve the group column's relation HERE (the only layer that
452        // knows the entity's metadata): a group on a relation FK gets its
453        // target table passed down so the inner query can carry the label.
454        let mut spec = spec.clone();
455        if let (Some(group), Some(label)) = (&spec.group_by, &spec.label_field) {
456            let _ = label;
457            let camel = snake_to_camel(group);
458            if let Some((_, table, _)) = T::relations().iter().find(|(_, _, fk)| *fk == camel) {
459                let base_fk = camel_to_snake(&camel);
460                spec.label_relation = Some((table.to_string(), base_fk));
461            }
462        }
463        self.inner
464            .aggregate_with_filters_and_base(
465                &spec,
466                base_condition,
467                filters,
468                column_types,
469                search_fields,
470            )
471            .await
472    }
473}
474
475// ─── SoftDelete mode ──────────────────────────────────────────────────────────
476
477impl<T> GenericCrudRepository<T, SoftDelete>
478where
479    T: for<'r> FromRow<'r, PgRow> + Send + Unpin,
480{
481    // ── Unique-field lookups (active records only) ────────────────────────────
482
483    /// Find an active entity by a unique text field.
484    ///
485    /// Filters `metadata->>'deleted_at' IS NULL` automatically.
486    pub async fn find_by_text_field(&self, field: &str, value: &str) -> Result<Option<T>> {
487        self.find_by_text_field_with_cond(field, value, " AND metadata->>'deleted_at' IS NULL").await
488    }
489
490    /// Check existence by a unique text field (active records only).
491    pub async fn exists_by_text_field(&self, field: &str, value: &str) -> Result<bool> {
492        self.exists_by_text_field_with_cond(field, value, " AND metadata->>'deleted_at' IS NULL").await
493    }
494
495    /// Find an active entity by a unique UUID field.
496    pub async fn find_by_uuid_field(&self, field: &str, value: Uuid) -> Result<Option<T>> {
497        self.find_by_uuid_field_with_cond(field, value, " AND metadata->>'deleted_at' IS NULL").await
498    }
499
500    /// Check existence by a unique UUID field (active records only).
501    pub async fn exists_by_uuid_field(&self, field: &str, value: Uuid) -> Result<bool> {
502        self.exists_by_uuid_field_with_cond(field, value, " AND metadata->>'deleted_at' IS NULL").await
503    }
504
505    // ── Filtered pagination (requires EntityRepoMeta on T) ────────────────────
506
507    /// Paginate active entities with filter and search support.
508    ///
509    /// Requires `T: EntityRepoMeta` for column type hints and search fields.
510    pub async fn list_paginated_filtered(
511        &self,
512        pagination: PaginationParams,
513        filters: Option<&HashMap<String, String>>,
514    ) -> Result<PaginatedResult<T>>
515    where
516        T: EntityRepoMeta + Send + Sync,
517    {
518        let filters_map = filters.cloned().unwrap_or_default();
519        let column_types = T::column_types();
520        let search_fields_owned: Vec<&'static str> = T::search_fields().iter().copied().collect();
521        self.run_filtered_query(
522            pagination,
523            Some("metadata->>'deleted_at' IS NULL"),
524            &filters_map,
525            &column_types,
526            &search_fields_owned,
527        ).await
528    }
529
530    /// Group and reduce active entities, skipping the soft-deleted.
531    pub async fn aggregate_filtered(
532        &self,
533        spec: &crate::repository::AggregateSpec,
534        filters: Option<&HashMap<String, String>>,
535    ) -> Result<crate::repository::AggregateResult>
536    where
537        T: EntityRepoMeta + Send + Sync,
538    {
539        let filters_map = filters.cloned().unwrap_or_default();
540        let column_types = T::column_types();
541        let search_fields_owned: Vec<&'static str> = T::search_fields().iter().copied().collect();
542        self.run_aggregate_query(spec, Some("metadata->>'deleted_at' IS NULL"), &filters_map, &column_types, &search_fields_owned).await
543    }
544
545    /// Paginate active entities, fenced to `company`.
546    ///
547    /// The company-scoped counterpart of [`Self::list_paginated_filtered`] and the one any
548    /// client-reachable read should use. Fails closed: a fenced entity with no company
549    /// returns [`MissingCompanyScope`] rather than every company's rows.
550    ///
551    /// The fence is ANDed into the same base condition the soft-delete guard uses, so
552    /// it lands in SQL — which is what keeps `total` honest. A post-filter in the handler
553    /// would fetch `limit` rows and return the survivors, leaving callers unable to tell
554    /// "end of data" from "filtered", and could not fix `COUNT` at all.
555    pub async fn list_paginated_filtered_scoped(
556        &self,
557        pagination: PaginationParams,
558        filters: Option<&HashMap<String, String>>,
559        company: Option<Uuid>,
560    ) -> Result<PaginatedResult<T>>
561    where
562        T: EntityRepoMeta + Send + Sync,
563    {
564        let fence = company_fence::<T>(company)?;
565        let mut filters_map = filters.cloned().unwrap_or_default();
566        strip_client_company_filters::<T>(&mut filters_map);
567        let column_types = T::column_types();
568        let search_fields_owned: Vec<&'static str> = T::search_fields().iter().copied().collect();
569        self.run_filtered_query(
570            pagination,
571            and_conditions(Some("metadata->>'deleted_at' IS NULL"), fence).as_deref(),
572            &filters_map,
573            &column_types,
574            &search_fields_owned,
575        )
576        .await
577    }
578
579    /// Paginate soft-deleted entities, fenced to `company`.
580    ///
581    /// `/trash` is the worse leak of the two: it serves rows whose owners were told the
582    /// data is gone. Same fence, same fail-closed contract.
583    pub async fn list_deleted_filtered_scoped(
584        &self,
585        pagination: PaginationParams,
586        filters: Option<&HashMap<String, String>>,
587        company: Option<Uuid>,
588    ) -> Result<PaginatedResult<T>>
589    where
590        T: EntityRepoMeta + Send + Sync,
591    {
592        let fence = company_fence::<T>(company)?;
593        let mut filters_map = filters.cloned().unwrap_or_default();
594        strip_client_company_filters::<T>(&mut filters_map);
595        let column_types = T::column_types();
596        let search_fields: Vec<&'static str> = T::search_fields().to_vec();
597        self.run_filtered_query(
598            pagination,
599            and_conditions(Some("metadata->>'deleted_at' IS NOT NULL"), fence).as_deref(),
600            &filters_map,
601            &column_types,
602            &search_fields,
603        )
604        .await
605    }
606
607    /// Paginate soft-deleted entities with filter and search support — the trash twin of
608    /// [`Self::list_paginated_filtered`].
609    pub async fn list_deleted_filtered(
610        &self,
611        pagination: PaginationParams,
612        filters: Option<&HashMap<String, String>>,
613    ) -> Result<PaginatedResult<T>>
614    where
615        T: EntityRepoMeta + Send + Sync,
616    {
617        let filters_map = filters.cloned().unwrap_or_default();
618        let column_types = T::column_types();
619        let search_fields: Vec<&'static str> = T::search_fields().to_vec();
620        self.run_filtered_query(
621            pagination,
622            Some("metadata->>'deleted_at' IS NOT NULL"),
623            &filters_map,
624            &column_types,
625            &search_fields,
626        ).await
627    }
628
629    /// Find an active (non-deleted) entity by primary key.
630    pub async fn find_by_id(&self, id: &str) -> Result<Option<T>> {
631        let query = format!(
632            "SELECT * FROM {} WHERE id = $1::uuid AND metadata->>'deleted_at' IS NULL",
633            self.table_name()
634        );
635        let result = crate::company_scope::fetch_optional_scoped(
636            self.pool(),
637            sqlx::query_as::<_, T>(&query).bind(id),
638        )
639        .await?;
640        Ok(result)
641    }
642
643    /// Return all active (non-deleted) entities.
644    pub async fn find_all(&self) -> Result<Vec<T>> {
645        let query = format!(
646            "SELECT * FROM {} WHERE metadata->>'deleted_at' IS NULL",
647            self.table_name()
648        );
649        let results = crate::company_scope::fetch_all_scoped(
650            self.pool(),
651            sqlx::query_as::<_, T>(&query),
652        )
653        .await?;
654        Ok(results)
655    }
656
657    /// Full update of an active record; `Ok(None)` when it is missing or soft-deleted.
658    ///
659    /// The not-deleted predicate is part of the UPDATE itself, so a write racing a soft
660    /// delete can never revive the row.
661    pub async fn update(&self, id: &str, entity: &T) -> Result<Option<T>>
662    where
663        T: Serialize + Send + Sync,
664    {
665        self.inner
666            .update_guarded(id, entity, None, ACTIVE_ROW_GUARD)
667            .await
668    }
669
670    /// Update an active record from `before` (the row as read) to `after`, writing only the
671    /// columns whose value differs between the two; `Ok(None)` when the row is missing or was
672    /// soft-deleted since it was read.
673    ///
674    /// The path for a read-modify-write such as a PATCH: a column the caller did not change is
675    /// left as it now stands in the database, so a concurrent write to it (a verb moving a
676    /// status) is not overwritten with the stale value read before it.
677    pub async fn update_changed(&self, id: &str, before: &T, after: &T) -> Result<Option<T>>
678    where
679        T: Serialize + Send + Sync,
680    {
681        let changed = crate::repository::changed_columns(before, after)?;
682        self.inner
683            .update_guarded(id, after, Some(&changed), ACTIVE_ROW_GUARD)
684            .await
685    }
686
687    /// Soft-delete an entity (sets `metadata.deleted_at`).
688    pub async fn delete(&self, id: &str) -> Result<bool> {
689        self.soft_delete(id).await
690    }
691
692    /// Count active (non-deleted) entities.
693    pub async fn count(&self) -> Result<u64> {
694        self.count_active().await
695    }
696
697    /// Return `true` if an active entity with the given ID exists.
698    pub async fn exists(&self, id: &str) -> Result<bool> {
699        let query = format!(
700            "SELECT 1 FROM {} WHERE id = $1::uuid AND metadata->>'deleted_at' IS NULL LIMIT 1",
701            self.table_name()
702        );
703        let result = crate::company_scope::fetch_optional_scalar_scoped(
704            self.pool(),
705            sqlx::query_scalar::<_, i32>(&query).bind(id),
706        )
707        .await?;
708        Ok(result.is_some())
709    }
710
711    /// Paginate active entities (most-recent-first by ID).
712    pub async fn list_paginated(&self, pagination: PaginationParams) -> Result<PaginatedResult<T>> {
713        let offset = pagination.offset();
714        let limit = pagination.limit();
715        let query = format!(
716            "SELECT * FROM {} WHERE metadata->>'deleted_at' IS NULL \
717             ORDER BY id DESC LIMIT $1 OFFSET $2",
718            self.table_name()
719        );
720        let data = crate::company_scope::fetch_all_scoped(
721            self.pool(),
722            sqlx::query_as::<_, T>(&query).bind(limit as i64).bind(offset as i64),
723        )
724        .await?;
725        let total = self.count_active().await?;
726        Ok(PaginatedResult {
727            data,
728            pagination: PaginationInfo::new(pagination.page, pagination.per_page, total),
729        })
730    }
731
732    // ── Soft-delete helpers ───────────────────────────────────────────────────
733
734    /// Set `metadata.deleted_at` to NOW() (soft delete).
735    pub async fn soft_delete(&self, id: &str) -> Result<bool> {
736        let query = format!(
737            "UPDATE {} SET metadata = jsonb_set(\
738               COALESCE(metadata, '{{}}'), \
739               '{{deleted_at}}', \
740               to_jsonb(NOW())\
741             ) WHERE id = $1::uuid AND (metadata->>'deleted_at') IS NULL",
742            self.table_name()
743        );
744        let result = crate::company_scope::execute_scoped(
745            self.pool(),
746            sqlx::query(&query).bind(id),
747        )
748        .await?;
749        Ok(result.rows_affected() > 0)
750    }
751
752    /// Remove `deleted_at` from metadata, restoring the entity.
753    pub async fn restore(&self, id: &str) -> Result<Option<T>> {
754        let query = format!(
755            "UPDATE {} SET metadata = metadata - 'deleted_at' \
756             WHERE id = $1::uuid AND (metadata->>'deleted_at') IS NOT NULL \
757             RETURNING *",
758            self.table_name()
759        );
760        let result = crate::company_scope::fetch_optional_scoped(
761            self.pool(),
762            sqlx::query_as::<_, T>(&query).bind(id),
763        )
764        .await?;
765        Ok(result)
766    }
767
768    /// Paginate soft-deleted entities (trash view).
769    pub async fn list_deleted(&self, pagination: PaginationParams) -> Result<PaginatedResult<T>> {
770        let offset = pagination.offset();
771        let limit = pagination.limit();
772        let query = format!(
773            "SELECT * FROM {} WHERE (metadata->>'deleted_at') IS NOT NULL \
774             ORDER BY (metadata->>'deleted_at') DESC LIMIT $1 OFFSET $2",
775            self.table_name()
776        );
777        let data = crate::company_scope::fetch_all_scoped(
778            self.pool(),
779            sqlx::query_as::<_, T>(&query).bind(limit as i64).bind(offset as i64),
780        )
781        .await?;
782        let count_query = format!(
783            "SELECT COUNT(*) FROM {} WHERE (metadata->>'deleted_at') IS NOT NULL",
784            self.table_name()
785        );
786        let total = crate::company_scope::fetch_one_scalar_scoped(
787            self.pool(),
788            sqlx::query_scalar::<_, i64>(&count_query),
789        )
790        .await? as u64;
791        Ok(PaginatedResult {
792            data,
793            pagination: PaginationInfo::new(pagination.page, pagination.per_page, total),
794        })
795    }
796
797    /// Permanently delete all soft-deleted rows (empty trash).
798    pub async fn empty_trash(&self) -> Result<u64> {
799        let query = format!(
800            "DELETE FROM {} WHERE (metadata->>'deleted_at') IS NOT NULL",
801            self.table_name()
802        );
803        let result = crate::company_scope::execute_scoped(self.pool(), sqlx::query(&query)).await?;
804        Ok(result.rows_affected())
805    }
806
807    /// Find a soft-deleted entity by primary key.
808    pub async fn find_deleted_by_id(&self, id: &str) -> Result<Option<T>> {
809        let query = format!(
810            "SELECT * FROM {} WHERE id = $1::uuid AND (metadata->>'deleted_at') IS NOT NULL",
811            self.table_name()
812        );
813        let result = crate::company_scope::fetch_optional_scoped(
814            self.pool(),
815            sqlx::query_as::<_, T>(&query).bind(id),
816        )
817        .await?;
818        Ok(result)
819    }
820
821    /// Permanently delete a soft-deleted entity by primary key.
822    pub async fn permanent_delete(&self, id: &str) -> Result<bool> {
823        let query = format!(
824            "DELETE FROM {} WHERE id = $1::uuid AND (metadata->>'deleted_at') IS NOT NULL",
825            self.table_name()
826        );
827        let result = crate::company_scope::execute_scoped(
828            self.pool(),
829            sqlx::query(&query).bind(id),
830        )
831        .await?;
832        Ok(result.rows_affected() > 0)
833    }
834
835    /// Count active (non-deleted) entities.
836    pub async fn count_active(&self) -> Result<u64> {
837        let query = format!(
838            "SELECT COUNT(*) FROM {} WHERE (metadata->>'deleted_at') IS NULL",
839            self.table_name()
840        );
841        let count = crate::company_scope::fetch_one_scalar_scoped(
842            self.pool(),
843            sqlx::query_scalar::<_, i64>(&query),
844        )
845        .await? as u64;
846        Ok(count)
847    }
848
849    /// Count soft-deleted entities.
850    pub async fn count_deleted(&self) -> Result<u64> {
851        let query = format!(
852            "SELECT COUNT(*) FROM {} WHERE (metadata->>'deleted_at') IS NOT NULL",
853            self.table_name()
854        );
855        let count = crate::company_scope::fetch_one_scalar_scoped(
856            self.pool(),
857            sqlx::query_scalar::<_, i64>(&query),
858        )
859        .await? as u64;
860        Ok(count)
861    }
862
863    // ── Atomic batch operations ───────────────────────────────────────────────
864    //
865    // Each runs inside a single transaction. For id-list operations the affected
866    // row count must equal the number of ids requested, otherwise the whole batch
867    // is rolled back — all-or-nothing semantics where a missing / already-in-the-
868    // target-state id (or a duplicate id) fails the entire request.
869
870    /// Soft-delete many active rows atomically.
871    pub async fn bulk_soft_delete(&self, ids: &[String]) -> Result<u64> {
872        if ids.is_empty() {
873            return Ok(0);
874        }
875        let placeholders = id_in_placeholders(ids.len());
876        let query = format!(
877            "UPDATE {} SET metadata = jsonb_set(\
878               COALESCE(metadata, '{{}}'), '{{deleted_at}}', to_jsonb(NOW())\
879             ) WHERE id IN ({placeholders}) AND (metadata->>'deleted_at') IS NULL",
880            self.table_name()
881        );
882        let mut tx = self.pool().begin().await?;
883        crate::company_scope::bind_current_company(&mut tx).await?;
884        let mut q = sqlx::query(&query);
885        for id in ids {
886            q = q.bind(id);
887        }
888        let affected = q.execute(&mut *tx).await?.rows_affected();
889        if affected != ids.len() as u64 {
890            // `tx` is dropped here without commit → rolled back.
891            return Err(anyhow::anyhow!(
892                "bulk_soft_delete: {} of {} ids were not active/deletable; rolled back",
893                ids.len() as u64 - affected,
894                ids.len()
895            ));
896        }
897        tx.commit().await?;
898        Ok(affected)
899    }
900
901    /// Restore many soft-deleted rows atomically, returning the restored rows.
902    pub async fn bulk_restore(&self, ids: &[String]) -> Result<Vec<T>> {
903        if ids.is_empty() {
904            return Ok(Vec::new());
905        }
906        let placeholders = id_in_placeholders(ids.len());
907        let query = format!(
908            "UPDATE {} SET metadata = metadata - 'deleted_at' \
909             WHERE id IN ({placeholders}) AND (metadata->>'deleted_at') IS NOT NULL \
910             RETURNING *",
911            self.table_name()
912        );
913        let mut tx = self.pool().begin().await?;
914        crate::company_scope::bind_current_company(&mut tx).await?;
915        let mut q = sqlx::query_as::<_, T>(&query);
916        for id in ids {
917            q = q.bind(id);
918        }
919        let rows = q.fetch_all(&mut *tx).await?;
920        if rows.len() != ids.len() {
921            return Err(anyhow::anyhow!(
922                "bulk_restore: {} of {} ids were not in trash; rolled back",
923                ids.len() - rows.len(),
924                ids.len()
925            ));
926        }
927        tx.commit().await?;
928        Ok(rows)
929    }
930
931    /// Permanently delete many soft-deleted rows atomically.
932    pub async fn bulk_permanent_delete(&self, ids: &[String]) -> Result<u64> {
933        if ids.is_empty() {
934            return Ok(0);
935        }
936        let placeholders = id_in_placeholders(ids.len());
937        let query = format!(
938            "DELETE FROM {} WHERE id IN ({placeholders}) AND (metadata->>'deleted_at') IS NOT NULL",
939            self.table_name()
940        );
941        let mut tx = self.pool().begin().await?;
942        crate::company_scope::bind_current_company(&mut tx).await?;
943        let mut q = sqlx::query(&query);
944        for id in ids {
945            q = q.bind(id);
946        }
947        let affected = q.execute(&mut *tx).await?.rows_affected();
948        if affected != ids.len() as u64 {
949            return Err(anyhow::anyhow!(
950                "bulk_permanent_delete: {} of {} ids were not in trash; rolled back",
951                ids.len() as u64 - affected,
952                ids.len()
953            ));
954        }
955        tx.commit().await?;
956        Ok(affected)
957    }
958
959    /// Restore every soft-deleted row, returning the restored rows. A single
960    /// `UPDATE ... RETURNING` is atomic, and the returned rows let the service
961    /// layer emit a `Restored` event per entity.
962    pub async fn restore_all(&self) -> Result<Vec<T>> {
963        let query = format!(
964            "UPDATE {} SET metadata = metadata - 'deleted_at' \
965             WHERE (metadata->>'deleted_at') IS NOT NULL \
966             RETURNING *",
967            self.table_name()
968        );
969        let rows = crate::company_scope::fetch_all_scoped(
970            self.pool(),
971            sqlx::query_as::<_, T>(&query),
972        )
973        .await?;
974        Ok(rows)
975    }
976
977    /// Update many active rows atomically. Every entity must reference an
978    /// existing active (non-soft-deleted) row or the whole batch is rolled back.
979    pub async fn bulk_update(&self, entities: &[T]) -> Result<Vec<T>>
980    where
981        T: Serialize + Send + Sync,
982    {
983        bulk_update_rows(
984            self.pool(),
985            self.table_name(),
986            ACTIVE_ROW_GUARD,
987            entities,
988        )
989        .await
990    }
991
992    /// Update many active rows from `(before, after)` pairs atomically, writing only the
993    /// columns each update changed; a row missing or soft-deleted since it was read rolls the
994    /// whole batch back.
995    pub async fn bulk_update_changed(&self, pairs: &[(T, T)]) -> Result<Vec<T>>
996    where
997        T: Serialize + Send + Sync,
998    {
999        self.bulk_update_changed_guarded(pairs, ACTIVE_ROW_GUARD).await
1000    }
1001}
1002
1003/// Build `"$1::uuid, $2::uuid, …"` for an `id IN (…)` clause of `n` bound ids.
1004fn id_in_placeholders(n: usize) -> String {
1005    (1..=n)
1006        .map(|i| format!("${i}::uuid"))
1007        .collect::<Vec<_>>()
1008        .join(", ")
1009}
1010
1011/// One item of a batch write failed, so the whole batch was rolled back.
1012///
1013/// `index` is the item's position in the batch as the caller passed it; `error` is why it
1014/// failed (a database refusal keeps its `sqlx::Error` inside, so a caller can still read the
1015/// constraint it broke).
1016#[derive(Debug)]
1017pub struct BatchItemError {
1018    /// Position of the failed item in the batch.
1019    pub index: usize,
1020    /// Why it failed.
1021    pub error: anyhow::Error,
1022}
1023
1024impl std::fmt::Display for BatchItemError {
1025    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
1026        write!(f, "item {} of the batch: {}; rolled back", self.index, self.error)
1027    }
1028}
1029
1030impl std::error::Error for BatchItemError {
1031    fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
1032        Some(self.error.as_ref())
1033    }
1034}
1035
1036/// Insert each entity on `conn` (an open transaction), stopping at the first failure.
1037async fn insert_each<T>(conn: &mut sqlx::PgConnection, table: &str, entities: &[T]) -> Result<Vec<T>>
1038where
1039    T: for<'r> FromRow<'r, PgRow> + Serialize + Send + Sync + Unpin,
1040{
1041    let mut out = Vec::with_capacity(entities.len());
1042    for (index, entity) in entities.iter().enumerate() {
1043        let (query, json) = crate::repository::insert_statement(table, entity)
1044            .map_err(|error| BatchItemError { index, error })?;
1045        let statement = match json {
1046            Some(json) => sqlx::query_as::<_, T>(&query).bind(json),
1047            None => sqlx::query_as::<_, T>(&query),
1048        };
1049        let row = statement.fetch_one(&mut *conn).await.map_err(|e| BatchItemError {
1050            index,
1051            error: crate::repository::explain_unknown_column(e, table),
1052        })?;
1053        out.push(row);
1054    }
1055    Ok(out)
1056}
1057
1058/// Apply each `(before, after)` pair on `conn` (an open transaction) as a changed-columns
1059/// update under `guard`, stopping at the first row that fails or no longer matches.
1060async fn update_each_changed<T>(
1061    conn: &mut sqlx::PgConnection,
1062    table: &str,
1063    guard: &str,
1064    pairs: &[(T, T)],
1065) -> Result<Vec<T>>
1066where
1067    T: for<'r> FromRow<'r, PgRow> + Serialize + Send + Sync + Unpin,
1068{
1069    use crate::repository::UpdateStatement;
1070    let mut out = Vec::with_capacity(pairs.len());
1071    for (index, (before, after)) in pairs.iter().enumerate() {
1072        let item = |error: anyhow::Error| BatchItemError { index, error };
1073        let id = match serde_json::to_value(after).map_err(|e| item(e.into()))? {
1074            serde_json::Value::Object(obj) => match obj.get("id") {
1075                Some(serde_json::Value::String(id)) => id.clone(),
1076                _ => return Err(item(anyhow::anyhow!("entity missing string 'id' field")).into()),
1077            },
1078            _ => return Err(item(anyhow::anyhow!("entity must serialize to a JSON object")).into()),
1079        };
1080        let changed = crate::repository::changed_columns(before, after).map_err(item)?;
1081        let updated = match crate::repository::update_statement(table, after, Some(&changed), guard)
1082            .map_err(item)?
1083        {
1084            UpdateStatement::Unchanged(query) => {
1085                sqlx::query_as::<_, T>(&query).bind(&id).fetch_optional(&mut *conn).await
1086            }
1087            UpdateStatement::Update { query, json } => {
1088                sqlx::query_as::<_, T>(&query)
1089                    .bind(json)
1090                    .bind(&id)
1091                    .fetch_optional(&mut *conn)
1092                    .await
1093            }
1094        }
1095        .map_err(|e| item(e.into()))?;
1096        match updated {
1097            Some(row) => out.push(row),
1098            None => {
1099                return Err(item(anyhow::anyhow!("id '{id}' not found or already deleted")).into())
1100            }
1101        }
1102    }
1103    Ok(out)
1104}
1105
1106/// Serialize an entity and return `(id, json_string, quoted_column_list)` for the
1107/// `jsonb_populate_record` update query — shared by the per-mode `bulk_update`s.
1108fn build_update_parts<T: Serialize>(entity: &T) -> Result<(String, String, String)> {
1109    let json_value = serde_json::to_value(entity)?;
1110    let json_obj = match json_value {
1111        serde_json::Value::Object(obj) => obj,
1112        _ => return Err(anyhow::anyhow!("entity must serialize to a JSON object")),
1113    };
1114    let id = json_obj
1115        .get("id")
1116        .and_then(|v| v.as_str())
1117        .ok_or_else(|| anyhow::anyhow!("entity missing string 'id' field"))?
1118        .to_string();
1119    let column_names = json_obj
1120        .keys()
1121        .filter(|k| *k != "id")
1122        .map(|k| format!("\"{k}\""))
1123        .collect::<Vec<_>>()
1124        .join(", ");
1125    let json_str = serde_json::to_string(&json_obj)?;
1126    Ok((id, json_str, column_names))
1127}
1128
1129// ─── HardDelete mode ─────────────────────────────────────────────────────────
1130
1131impl<T> GenericCrudRepository<T, HardDelete>
1132where
1133    T: for<'r> FromRow<'r, PgRow> + Send + Sync + Unpin + Serialize,
1134{
1135    // ── Unique-field lookups (no soft-delete guard) ───────────────────────────
1136
1137    /// Find an entity by a unique text field (no soft-delete guard).
1138    pub async fn find_by_text_field(&self, field: &str, value: &str) -> Result<Option<T>> {
1139        self.find_by_text_field_with_cond(field, value, "").await
1140    }
1141
1142    /// Check existence by a unique text field.
1143    pub async fn exists_by_text_field(&self, field: &str, value: &str) -> Result<bool> {
1144        self.exists_by_text_field_with_cond(field, value, "").await
1145    }
1146
1147    /// Find an entity by a unique UUID field.
1148    pub async fn find_by_uuid_field(&self, field: &str, value: Uuid) -> Result<Option<T>> {
1149        self.find_by_uuid_field_with_cond(field, value, "").await
1150    }
1151
1152    /// Check existence by a unique UUID field.
1153    pub async fn exists_by_uuid_field(&self, field: &str, value: Uuid) -> Result<bool> {
1154        self.exists_by_uuid_field_with_cond(field, value, "").await
1155    }
1156
1157    // ── Filtered pagination ───────────────────────────────────────────────────
1158
1159    /// Paginate entities with filter and search support.
1160    pub async fn list_paginated_filtered(
1161        &self,
1162        pagination: PaginationParams,
1163        filters: Option<&HashMap<String, String>>,
1164    ) -> Result<PaginatedResult<T>>
1165    where
1166        T: EntityRepoMeta + Send + Sync,
1167    {
1168        let filters_map = filters.cloned().unwrap_or_default();
1169        let column_types = T::column_types();
1170        let search_fields_owned: Vec<&'static str> = T::search_fields().iter().copied().collect();
1171        self.run_filtered_query(pagination, None, &filters_map, &column_types, &search_fields_owned).await
1172    }
1173
1174    /// Group and reduce entities. This mode has no trash to exclude.
1175    pub async fn aggregate_filtered(
1176        &self,
1177        spec: &crate::repository::AggregateSpec,
1178        filters: Option<&HashMap<String, String>>,
1179    ) -> Result<crate::repository::AggregateResult>
1180    where
1181        T: EntityRepoMeta + Send + Sync,
1182    {
1183        let filters_map = filters.cloned().unwrap_or_default();
1184        let column_types = T::column_types();
1185        let search_fields_owned: Vec<&'static str> = T::search_fields().iter().copied().collect();
1186        self.run_aggregate_query(spec, None, &filters_map, &column_types, &search_fields_owned).await
1187    }
1188
1189    /// Find an entity by primary key.
1190    pub async fn find_by_id(&self, id: &str) -> Result<Option<T>> {
1191        self.inner.find_by_id(id).await
1192    }
1193
1194    /// Return all entities.
1195    pub async fn find_all(&self) -> Result<Vec<T>> {
1196        let query = format!("SELECT * FROM {}", self.table_name());
1197        let results = crate::company_scope::fetch_all_scoped(
1198            self.pool(),
1199            sqlx::query_as::<_, T>(&query),
1200        )
1201        .await?;
1202        Ok(results)
1203    }
1204
1205    /// Full update.
1206    pub async fn update(&self, id: &str, entity: &T) -> Result<Option<T>> {
1207        self.inner.update(id, entity).await
1208    }
1209
1210    /// Update a record from `before` (the row as read) to `after`, writing only the columns
1211    /// whose value differs; `Ok(None)` when the row no longer exists. See the soft-delete
1212    /// twin for why a read-modify-write must not write back the columns it did not change.
1213    pub async fn update_changed(&self, id: &str, before: &T, after: &T) -> Result<Option<T>> {
1214        let changed = crate::repository::changed_columns(before, after)?;
1215        self.inner
1216            .update_guarded(id, after, Some(&changed), "")
1217            .await
1218    }
1219
1220    /// Permanently delete an entity by primary key.
1221    pub async fn delete(&self, id: &str) -> Result<bool> {
1222        self.inner.delete(id).await
1223    }
1224
1225    /// Count all entities.
1226    pub async fn count(&self) -> Result<u64> {
1227        let query = format!("SELECT COUNT(*) FROM {}", self.table_name());
1228        let count = crate::company_scope::fetch_one_scalar_scoped(
1229            self.pool(),
1230            sqlx::query_scalar::<_, i64>(&query),
1231        )
1232        .await? as u64;
1233        Ok(count)
1234    }
1235
1236    /// Return `true` if an entity with the given ID exists.
1237    pub async fn exists(&self, id: &str) -> Result<bool> {
1238        let query = format!(
1239            "SELECT 1 FROM {} WHERE id = $1::uuid LIMIT 1",
1240            self.table_name()
1241        );
1242        let result = crate::company_scope::fetch_optional_scalar_scoped(
1243            self.pool(),
1244            sqlx::query_scalar::<_, i32>(&query).bind(id),
1245        )
1246        .await?;
1247        Ok(result.is_some())
1248    }
1249
1250    /// Paginate entities (most-recent-first by ID).
1251    pub async fn list_paginated(&self, pagination: PaginationParams) -> Result<PaginatedResult<T>> {
1252        let offset = pagination.offset();
1253        let limit = pagination.limit();
1254        let query = format!(
1255            "SELECT * FROM {} ORDER BY id DESC LIMIT $1 OFFSET $2",
1256            self.table_name()
1257        );
1258        let data = crate::company_scope::fetch_all_scoped(
1259            self.pool(),
1260            sqlx::query_as::<_, T>(&query).bind(limit as i64).bind(offset as i64),
1261        )
1262        .await?;
1263        let total = self.count().await?;
1264        Ok(PaginatedResult {
1265            data,
1266            pagination: PaginationInfo::new(pagination.page, pagination.per_page, total),
1267        })
1268    }
1269
1270    // ── Atomic batch operations ───────────────────────────────────────────────
1271
1272    /// Hard-delete many rows atomically. Rolls back unless every id matched.
1273    pub async fn bulk_delete(&self, ids: &[String]) -> Result<u64> {
1274        if ids.is_empty() {
1275            return Ok(0);
1276        }
1277        let placeholders = id_in_placeholders(ids.len());
1278        let query = format!(
1279            "DELETE FROM {} WHERE id IN ({placeholders})",
1280            self.table_name()
1281        );
1282        let mut tx = self.pool().begin().await?;
1283        crate::company_scope::bind_current_company(&mut tx).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 *tx).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        tx.commit().await?;
1297        Ok(affected)
1298    }
1299
1300    /// Update many rows atomically. No soft-delete guard in this mode.
1301    pub async fn bulk_update(&self, entities: &[T]) -> Result<Vec<T>> {
1302        bulk_update_rows(self.pool(), self.table_name(), "", entities).await
1303    }
1304
1305    /// Update many rows from `(before, after)` pairs atomically, writing only the columns each
1306    /// update changed; a row deleted since it was read rolls the whole batch back.
1307    pub async fn bulk_update_changed(&self, pairs: &[(T, T)]) -> Result<Vec<T>> {
1308        self.bulk_update_changed_guarded(pairs, "").await
1309    }
1310}
1311
1312/// Qualify a relation target table with the calling entity's schema.
1313///
1314/// `EntityRepoMeta::relations()` emits bare table names (e.g. `countries`), but
1315/// module tables live in the module's own schema (`geo.countries`) and the
1316/// runtime app role's search_path does not include module schemas — an
1317/// unqualified reference fails with "relation does not exist". The calling
1318/// entity's table name is schema-qualified, so its prefix is the schema a
1319/// relation declared inside one module shares with its target. Targets that
1320/// already carry a schema, and callers that do not, pass through unchanged.
1321pub fn qualify_relation_table(caller_table: &str, target_table: &str) -> String {
1322    if target_table.contains('.') || !caller_table.contains('.') {
1323        target_table.to_string()
1324    } else {
1325        let schema = caller_table.split('.').next().unwrap_or(caller_table);
1326        format!("{schema}.{target_table}")
1327    }
1328}
1329
1330/// True when the error is Postgres `undefined_table` (SQLSTATE 42P01).
1331///
1332/// Walks the anyhow chain because the scoped-fetch error is converted with `?`
1333/// on its way up — the sqlx error rides as the (only) source.
1334fn is_undefined_table(err: &anyhow::Error) -> bool {
1335    err.chain()
1336        .filter_map(|cause| cause.downcast_ref::<sqlx::Error>())
1337        .any(|sqlx_err| {
1338            sqlx_err
1339                .as_database_error()
1340                .map(|db| db.code().as_deref() == Some("42P01"))
1341                .unwrap_or(false)
1342        })
1343}
1344
1345/// Fetch rows from an arbitrary table by id list, as JSON — hydrates `?include=`
1346/// relations in the generic CRUD handler without a typed repository for the
1347/// target entity. `table` is a generator-emitted collection name (from
1348/// `EntityRepoMeta::relations()`), NEVER client input, so the interpolation is
1349/// not an injection vector. `caller_table` is the hydrated entity's own
1350/// (schema-qualified) table name, used to qualify a bare `table` — see
1351/// [`qualify_relation_table`]. One batched `WHERE id = ANY(...)` per relation.
1352/// Returns `row_to_json` objects keyed by raw (snake_case) column names.
1353///
1354/// Resolution is two-step for bare targets: the caller's schema first, then —
1355/// only on `undefined_table` — the name as written, which resolves through the
1356/// role's search_path (platform tables such as `users` or `roles` live in
1357/// `public` while their declaring entities live in a module schema). Any other
1358/// error, or a target that fails both spellings, is returned to the caller.
1359pub async fn fetch_by_ids_as_json(
1360    pool: &PgPool,
1361    caller_table: &str,
1362    table: &str,
1363    ids: &[String],
1364) -> Result<Vec<serde_json::Value>> {
1365    if ids.is_empty() {
1366        return Ok(Vec::new());
1367    }
1368    let qualified = qualify_relation_table(caller_table, table);
1369    match fetch_rows_as_json(pool, &qualified, ids).await {
1370        Ok(rows) => Ok(rows),
1371        Err(err) if qualified != table && is_undefined_table(&err) => {
1372            fetch_rows_as_json(pool, table, ids).await
1373        }
1374        Err(err) => Err(err),
1375    }
1376}
1377
1378/// The scoped fetch behind [`fetch_by_ids_as_json`] — one relation, one table
1379/// spelling. Scoped because an `?include=` hydration of a company-fenced
1380/// relation must obey the same fence, so a cross-company related row is never
1381/// leaked through the expansion.
1382async fn fetch_rows_as_json(
1383    pool: &PgPool,
1384    table: &str,
1385    ids: &[String],
1386) -> Result<Vec<serde_json::Value>> {
1387    let query = format!("SELECT row_to_json(t) AS j FROM {table} t WHERE t.id = ANY($1::uuid[])");
1388    let rows: Vec<(serde_json::Value,)> =
1389        crate::company_scope::fetch_all_scoped(pool, sqlx::query_as(&query).bind(ids)).await?;
1390    Ok(rows.into_iter().map(|(j,)| j).collect())
1391}
1392
1393/// Shared transactional bulk-update used by both delete modes. Each entity is
1394/// updated by id inside one transaction; a missing (or, with `active_guard`,
1395/// soft-deleted) row rolls the whole batch back. `active_guard` is an extra SQL
1396/// predicate ANDed into the `WHERE` clause (`""` for none).
1397async fn bulk_update_rows<T>(
1398    pool: &PgPool,
1399    table: &str,
1400    active_guard: &str,
1401    entities: &[T],
1402) -> Result<Vec<T>>
1403where
1404    T: for<'r> FromRow<'r, PgRow> + Send + Sync + Unpin + Serialize,
1405{
1406    if entities.is_empty() {
1407        return Ok(Vec::new());
1408    }
1409    let mut tx = pool.begin().await?;
1410    crate::company_scope::bind_current_company(&mut tx).await?;
1411    let mut out = Vec::with_capacity(entities.len());
1412    for entity in entities {
1413        let (id, json_str, column_names) = build_update_parts(entity)?;
1414        let query = format!(
1415            "WITH new_row AS (\
1416                SELECT (jsonb_populate_record(NULL::{table}, $1::jsonb)).*\
1417             ) UPDATE {table} AS t \
1418             SET ({columns}) = (SELECT {columns} FROM new_row) \
1419             WHERE t.id = $2::uuid{guard} \
1420             RETURNING t.*",
1421            table = table,
1422            columns = column_names,
1423            guard = active_guard,
1424        );
1425        let updated = sqlx::query_as::<_, T>(&query)
1426            .bind(&json_str)
1427            .bind(&id)
1428            .fetch_optional(&mut *tx)
1429            .await?;
1430        match updated {
1431            Some(e) => out.push(e),
1432            None => {
1433                return Err(anyhow::anyhow!(
1434                    "bulk_update: id '{id}' not found or already deleted; rolled back"
1435                ));
1436            }
1437        }
1438    }
1439    tx.commit().await?;
1440    Ok(out)
1441}
1442
1443/// camelCase → snake_case (`projectId` → `project_id`), the SQL vocabulary.
1444fn camel_to_snake(s: &str) -> String {
1445    let mut out = String::with_capacity(s.len() + 4);
1446    for ch in s.chars() {
1447        if ch.is_uppercase() {
1448            out.push('_');
1449            out.extend(ch.to_lowercase());
1450        } else {
1451            out.push(ch);
1452        }
1453    }
1454    out
1455}