backbone_orm/

repository.rs

//! Repository implementations for PostgreSQL with comprehensive CRUD operations

use async_trait::async_trait;
use sqlx::{PgPool, FromRow, postgres::PgRow, Postgres};
use serde::{Deserialize, Serialize};
use serde_json::Value;
use chrono::NaiveDateTime;
use std::collections::{HashMap, HashSet};

use crate::qualify_relation_table;
use crate::filter::{parse_filters as parse_query_filter};
use crate::filter::SortDirection as FilterSortDirection;
use sqlx::Row as _;

/// Generic entity trait that all repository entities must implement
pub trait Entity {
    /// Get the entity's ID
    fn id(&self) -> Option<&str>;

    /// Get the table name for this entity
    fn table_name() -> &'static str where Self: Sized;

    /// Check if entity is soft deleted
    fn is_deleted(&self) -> bool { false }

    /// Get creation timestamp
    fn created_at(&self) -> Option<NaiveDateTime> { None }

    /// Get update timestamp
    fn updated_at(&self) -> Option<NaiveDateTime> { None }
}

/// Pagination parameters
#[derive(Debug, Clone, Default)]
pub struct PaginationParams {
    pub page: u32,
    pub per_page: u32,
}

impl PaginationParams {
    /// The most rows one page may hold. Every list the generic repositories serve is capped
    /// here, whatever page size the caller asked for.
    pub const MAX_PER_PAGE: u32 = 100;

    /// A page request, normalized: page 0 reads as page 1, and the page size is held to
    /// `1..=`[`MAX_PER_PAGE`](Self::MAX_PER_PAGE). Build pagination through this rather than
    /// a struct literal, which skips both.
    pub fn new(page: u32, per_page: u32) -> Self {
        Self {
            page: page.max(1),
            per_page: per_page.clamp(1, Self::MAX_PER_PAGE),
        }
    }

    /// The rows to skip. Page 0 counts as page 1, so a literal that skipped [`Self::new`]
    /// still never underflows.
    pub fn offset(&self) -> u32 {
        self.page.saturating_sub(1).saturating_mul(self.per_page)
    }

    pub fn limit(&self) -> u32 {
        self.per_page
    }
}

/// Sorting parameters
#[derive(Debug, Clone, Default)]
pub struct SortParams {
    pub field: String,
    pub direction: SortDirection,
}

#[derive(Debug, Clone, Default)]
pub enum SortDirection {
    #[default]
    Asc,
    Desc,
}

/// Filter parameters
#[derive(Debug, Clone, Default)]
pub struct FilterParams {
    pub conditions: HashMap<String, FilterCondition>,
}

#[derive(Debug, Clone)]
pub enum FilterCondition {
    Equals(String),
    NotEquals(String),
    GreaterThan(String),
    LessThan(String),
    Like(String),
    In(Vec<String>),
    IsNull,
    IsNotNull,
}

/// Paginated result wrapper
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct PaginatedResult<T> {
    pub data: Vec<T>,
    pub pagination: PaginationInfo,
}

#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct PaginationInfo {
    pub page: u32,
    pub per_page: u32,
    pub total: u64,
    pub total_pages: u32,
    /// Keyset paging: the position of this page's last row, for the next
    /// page's `after=`. None when the page is empty or no more rows follow.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub next_cursor: Option<String>,
    /// Keyset paging: the position of this page's first row, for the
    /// previous page's `before=`.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub prev_cursor: Option<String>,
    /// Whether another page follows (fetched with limit+1, so it is known
    /// without a count).
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub has_more: Option<bool>,
    /// How `total` came to be: "exact" (counted), "estimate" (the planner's
    /// row estimate — `estimate=1` asked for it), or "none" (a cursor walk
    /// without an estimate; the exact figure is the separate count call).
    #[serde(default = "default_count_mode")]
    pub count_mode: String,
}

/// Below this many rows (by the planner's estimate) an estimated total is counted exactly
/// instead: the count is cheap, and a planner estimate is least reliable on small sets.
pub const EXACT_COUNT_BELOW: u64 = 10_000;

fn default_count_mode() -> String {
    "exact".to_string()
}

impl PaginationInfo {
    pub fn new(page: u32, per_page: u32, total: u64) -> Self {
        let total_pages = ((total as f64) / (per_page as f64)).ceil() as u32;
        Self {
            page,
            per_page,
            total,
            total_pages,
            next_cursor: None,
            prev_cursor: None,
            has_more: None,
            count_mode: default_count_mode(),
        }
    }
}

/// Database operations trait - requires Serialize for write operations
#[async_trait]
pub trait DatabaseOperations<T: for<'a> FromRow<'a, PgRow> + Send + Unpin> {
    /// Create a new entity
    async fn create(&self, entity: &T) -> anyhow::Result<T>;

    /// Find entity by ID
    async fn find_by_id(&self, id: &str) -> anyhow::Result<Option<T>>;

    /// Find all entities
    async fn find_all(&self) -> anyhow::Result<Vec<T>>;

    /// Update an existing entity
    async fn update(&self, id: &str, entity: &T) -> anyhow::Result<Option<T>>;

    /// Delete an entity
    async fn delete(&self, id: &str) -> anyhow::Result<bool>;

    /// Count all entities
    async fn count(&self) -> anyhow::Result<u64>;

    /// Check if entity exists
    async fn exists(&self, id: &str) -> anyhow::Result<bool>;

    /// Execute custom query
    async fn execute_query(&self, query: &str) -> anyhow::Result<u64>;
}

/// PostgreSQL repository implementation with JSON-based dynamic queries
pub struct PostgresRepository<T: for<'a> FromRow<'a, PgRow> + Send + Unpin> {
    pool: PgPool,
    table_name: String,
    _phantom: std::marker::PhantomData<T>,
}

impl<T: for<'a> FromRow<'a, PgRow> + Send + Unpin> PostgresRepository<T> {
    pub fn new(pool: PgPool, table_name: &str) -> Self {
        Self {
            pool,
            table_name: table_name.to_string(),
            _phantom: std::marker::PhantomData,
        }
    }

    pub fn pool(&self) -> &PgPool {
        &self.pool
    }

    pub fn table_name(&self) -> &str {
        &self.table_name
    }

    /// List entities with pagination and advanced filtering
    ///
    /// This method provides comprehensive filtering capabilities similar to Laravel's Filter Query String.
    ///
    /// # Supported Filter Operators
    ///
    /// - `field[eq]=value` - Equal
    /// - `field[notEq]=value` - Not equal
    /// - `field[gt]=value` - Greater than
    /// - `field[gte]=value` - Greater than or equal
    /// - `field[lt]=value` - Less than
    /// - `field[lte]=value` - Less than or equal
    /// - `field[like]=value` - LIKE (case-sensitive)
    /// - `field[ilike]=value` - ILIKE (case-insensitive)
    /// - `field[notlike]=value` - NOT LIKE
    /// - `field[contain]=value` - Contains (%value%)
    /// - `field[notcontain]=value` - Does not contain
    /// - `field[startwith]=value` - Starts with (value%)
    /// - `field[endwith]=value` - Ends with (%value)
    /// - `field[in]=val1,val2` - IN array
    /// - `field[notin]=val1,val2` - NOT IN array
    /// - `field[between]=val1,val2` - BETWEEN
    /// - `field[notbetween]=val1,val2` - NOT BETWEEN
    /// - `field[isnull]` - IS NULL
    /// - `field[isnotnull]` - IS NOT NULL
    ///
    /// # Special Parameters
    ///
    /// - `search=value&searchFields=field1,field2` - Search in multiple fields
    /// - `orderby=field` or `orderby[field]=asc` - Sort results
    /// - `limit=10` - Limit results
    /// - `page=1` - Page number
    ///
    /// # Column Type Casting
    ///
    /// The `column_types` HashMap maps field names to their PostgreSQL types for proper casting.
    /// For example, `{"status": "user_status"}` will cast the status parameter to `user_status` enum type.
    ///
    /// # Example
    ///
    /// ```ignore
    /// let mut filters = HashMap::new();
    /// filters.insert("username[contain]".to_string(), "john".to_string());
    /// filters.insert("age[gt]".to_string(), "18".to_string());
    ///
    /// let mut column_types = HashMap::new();
    /// column_types.insert("status".to_string(), "user_status".to_string());
    ///
    /// let result = repo.list_with_filters(
    ///     PaginationParams::new(1, 10),
    ///     &filters,
    ///     &column_types,
    ///     &["username", "email"]  // search fields
    /// ).await?;
    /// ```
    pub async fn list_with_filters(
        &self,
        pagination: PaginationParams,
        filters: &HashMap<String, String>,
        column_types: &HashMap<String, String>,
        search_fields: &[&str],
    ) -> anyhow::Result<PaginatedResult<T>>
    where
        T: Send + Sync,
    {
        self.list_with_filters_and_base(pagination, None, filters, column_types, search_fields)
            .await
    }

    /// [`Self::list_with_filters`] under a repository-authored base condition.
    ///
    /// `base_condition` is raw SQL ANDed, as its own parenthesised group, ahead of everything
    /// the request asked for. It must come from code (a soft-delete guard, the company fence),
    /// never from the request — which is why it is a separate argument and not a filter key:
    /// the wire parser ignores every `__`-prefixed key.
    pub(crate) async fn list_with_filters_and_base(
        &self,
        pagination: PaginationParams,
        base_condition: Option<&str>,
        filters: &HashMap<String, String>,
        column_types: &HashMap<String, String>,
        search_fields: &[&str],
    ) -> anyhow::Result<PaginatedResult<T>>
    where
        T: Send + Sync,
    {
        // A parameter the request got wrong is refused by name, whether the parser or the
        // database found it out.
        let listed = async {
            // Parse filters from HashMap (no field allow-list by default for backward compatibility)
            let mut query_filter = self.parse_typed_filters(filters, column_types, None).await?;
            if let Some(cond) = base_condition {
                query_filter.add_base_condition(cond);
            }

            // Set up search fields if provided
            if !search_fields.is_empty() {
                query_filter.search_fields = search_fields.iter().map(|s| s.to_string()).collect();
            }

            self.execute_list(pagination, query_filter).await
        }
        .await;
        match listed {
            Ok(page) => Ok(page),
            Err(e) => Err(self.name_refused_parameter(e, filters).await),
        }
    }

    /// List entities with pagination, filtering, and field whitelist enforcement
    ///
    /// Similar to `list_with_filters` but accepts an optional set of allowed field names.
    /// When provided, only filter conditions on whitelisted fields are applied;
    /// conditions on unknown fields are silently dropped.
    ///
    /// This prevents clients from filtering on internal or sensitive columns
    /// (e.g., `password_hash`, `internal_notes`).
    ///
    /// # Arguments
    ///
    /// * `pagination` - Page and limit parameters
    /// * `filters` - HTTP query parameters (e.g., `field[operator]=value`)
    /// * `column_types` - PostgreSQL type mappings for enum casting
    /// * `search_fields` - Fields to search when `search` parameter is present
    /// * `allowed_fields` - Optional whitelist of field names; `None` allows all fields
    ///
    /// # Example
    ///
    /// ```ignore
    /// let allowed: HashSet<String> = ["username", "email", "status"]
    ///     .iter().map(|s| s.to_string()).collect();
    ///
    /// let result = repo.list_with_filters_whitelisted(
    ///     PaginationParams::new(1, 10),
    ///     &filters,
    ///     &column_types,
    ///     &["username", "email"],
    ///     Some(&allowed),
    /// ).await?;
    /// ```
    pub async fn list_with_filters_whitelisted(
        &self,
        pagination: PaginationParams,
        filters: &HashMap<String, String>,
        column_types: &HashMap<String, String>,
        search_fields: &[&str],
        allowed_fields: Option<&HashSet<String>>,
    ) -> anyhow::Result<PaginatedResult<T>>
    where
        T: Send + Sync,
    {
        let listed = async {
            // Parse filters with optional field whitelist
            let mut query_filter =
                self.parse_typed_filters(filters, column_types, allowed_fields).await?;

            // Set up search fields if provided
            if !search_fields.is_empty() {
                query_filter.search_fields = search_fields.iter().map(|s| s.to_string()).collect();
            }

            self.execute_list(pagination, query_filter).await
        }
        .await;
        match listed {
            Ok(page) => Ok(page),
            Err(e) => Err(self.name_refused_parameter(e, filters).await),
        }
    }

    /// Update one row by id, setting only `only` columns when given (every serialized column but
    /// `id` otherwise), and only while `guard` holds.
    ///
    /// `guard` is repository-authored SQL ANDed into the `WHERE` after the id match (`""` for
    /// none), written against the table alias `t` — e.g. `" AND t.metadata->>'deleted_at' IS
    /// NULL"`. Putting the not-deleted predicate in the UPDATE itself is what keeps a write that
    /// races a soft delete from reviving the row: a separate existence check before the UPDATE
    /// leaves a window in which the delete lands. `Ok(None)` when no row matched.
    ///
    /// With `only` empty nothing is written, and the current row (under the same guard) is
    /// returned unchanged.
    pub async fn update_guarded(
        &self,
        id: &str,
        entity: &T,
        only: Option<&[String]>,
        guard: &str,
    ) -> anyhow::Result<Option<T>>
    where
        T: Serialize + Send + Sync,
    {
        let result = match update_statement(&self.table_name, entity, only, guard)? {
            UpdateStatement::Unchanged(query) => {
                crate::company_scope::fetch_optional_scoped(
                    &self.pool,
                    sqlx::query_as::<_, T>(&query).bind(id),
                )
                .await?
            }
            UpdateStatement::Update { query, json } => {
                crate::company_scope::fetch_optional_scoped(
                    &self.pool,
                    sqlx::query_as::<_, T>(&query).bind(json).bind(id),
                )
                .await?
            }
        };

        Ok(result)
    }

    /// Parse the wire filters with a cast for every typed column they compare.
    ///
    /// Filter values arrive as text and are bound as text; PostgreSQL has no implicit comparison
    /// between text and a boolean, number, uuid, date, time or timestamp column, so each such
    /// comparison needs a cast on its placeholder. The entity's generated `column_types()` hints
    /// supply it where they exist, and stay the answer for the columns they name. They were never
    /// a complete list — no booleans or numbers, only uuids named `id`/`*_id`, and temporal
    /// columns only in modules generated after the generator learned them — so whenever a
    /// comparison is left without a cast, the table's real column types are read from the
    /// catalog and fill the gaps. The catalog cannot drift from the table the query runs
    /// against, which is the same reason the aggregate and sort paths read it.
    ///
    /// A filter on text columns only, or with hints for every compared column, costs nothing
    /// extra beyond the parse; otherwise one catalog lookup on the request's own connection.
    async fn parse_typed_filters(
        &self,
        filters: &HashMap<String, String>,
        column_types: &HashMap<String, String>,
        allowed_fields: Option<&HashSet<String>>,
    ) -> anyhow::Result<crate::QueryFilter> {
        let query_filter = parse_query_filter(filters, column_types, allowed_fields)?;
        if !query_filter.needs_catalog_types() {
            return Ok(query_filter);
        }
        let catalog = catalog_filter_casts(&self.pool, &self.table_name).await?;
        let merged = merge_filter_casts(column_types, catalog);
        parse_query_filter(filters, &merged, allowed_fields)
    }

    /// The shared list execution: filters, order, paging, and the total.
    ///
    /// Three shapes, chosen by the request:
    ///
    /// * **page mode** (today's behaviour, unchanged): exact COUNT, then
    ///   `LIMIT l OFFSET o` in the requested order;
    /// * **cursor mode** (`after=`/`before=`): the keyset predicate replaces
    ///   the offset, the order always ends on the `id` tiebreaker, and one
    ///   extra row is fetched so `has_more` is known without a count — the
    ///   page costs the same at any depth, which is the point;
    /// * **estimate** (`estimate=1`, either mode): the exact COUNT (a scan
    ///   of the whole filtered set) is replaced by the planner's row
    ///   estimate; the exact figure stays the separate count call.
    #[allow(clippy::type_complexity)]
    async fn execute_list(
        &self,
        pagination: PaginationParams,
        query_filter: crate::QueryFilter,
    ) -> anyhow::Result<PaginatedResult<T>> {
        let limit = pagination.limit() as i64;
        let backwards =
            query_filter.cursor_before.is_some() && query_filter.cursor_after.is_none();
        let cursor_walk = query_filter.cursor_after.is_some() || backwards;

        let (mut where_clause, mut filter_params) = query_filter.build_where_clause();
        let order_clause;
        // Cast suffixes for the sort columns, for the keyset binds.
        let mut boundary_casts: Vec<Option<String>> = Vec::new();

        // The total order every paged read carries — see `total_order`. A
        // cursor walks it, and a page-mode list gets the same order so it can
        // HAND OUT a cursor to start a keyset walk from, and so offset paging
        // returns each row exactly once whatever the physical row order does
        // between two page fetches.
        let sorts = total_order(
            query_filter
                .sorts
                .iter()
                .map(|s| (s.field.clone(), s.direction.clone()))
                .collect(),
        );

        // Casts for the sort columns: cursor mode walks on them, and a page
        // that sorts by something other than the key encodes its handed-out
        // cursor from them. A list ordered by the key alone needs none —
        // the id decodes as a uuid on its own — which spares the plain
        // unsorted list a catalog round trip it never needed.
        if cursor_walk || sorts.iter().any(|(f, _)| f != "id") {
            boundary_casts = self.sort_column_casts(&sorts).await?;
        }

        if cursor_walk {
            let opaque = if backwards {
                query_filter.cursor_before.clone().unwrap()
            } else {
                query_filter.cursor_after.clone().unwrap()
            };
            let payload = crate::filter::cursor::decode_cursor(&opaque, &sorts).map_err(|e| {
                // The parameter it arrived in is `before` only when `after` was absent (see
                // `backwards`).
                let parameter = if backwards { "before" } else { "after" };
                let why = match e {
                    crate::filter::CursorError::SortMismatch { .. } => {
                        "it was issued for a different sort order"
                    }
                    crate::filter::CursorError::Version(_) => "it was issued by an older version",
                    _ => "it is not a position this list issued",
                };
                anyhow::Error::from(crate::filter::QueryParamError::new(
                    parameter,
                    crate::filter::QueryParamReason::InvalidCursor,
                    format!("`{parameter}` cannot continue this list: {why}"),
                ))
            })?;
            let mut idx = filter_params.len() + 1;
            let (keyset_sql, keyset_params) = crate::filter::cursor::build_keyset_predicate(
                &payload,
                &mut idx,
                &boundary_casts,
                backwards,
            );
            where_clause = and_keyset_predicate(&where_clause, &keyset_sql);
            filter_params.extend(keyset_params);
            order_clause = order_by_sql(&sorts, backwards);
        } else {
            // Page mode: the caller's order (or the key's), already total.
            order_clause = order_by_sql(&sorts, false);
        }
        // The deterministic order the handed-out cursors are cut in.
        let boundary_sorts: Vec<(String, FilterSortDirection)> = sorts;

        // The total: exact in page mode (today's behaviour), the planner's
        // estimate when asked, nothing on a cursor walk that did not ask.
        // Asked for an estimate, a set the planner expects to be small is
        // counted all the same: the count is cheap there and the estimate is
        // at its least reliable, so "about 12" never stands in for 10.
        let (total, count_mode) = if query_filter.estimate_total {
            let estimate = self.estimate_filtered_rows(&where_clause, &filter_params).await?;
            if estimate <= EXACT_COUNT_BELOW {
                (self.count_filtered_rows(&where_clause, &filter_params).await?, "exact")
            } else {
                (estimate, "estimate")
            }
        } else if cursor_walk {
            (0u64, "none")
        } else {
            (self.count_filtered_rows(&where_clause, &filter_params).await?, "exact")
        };

        // The page: limit+1 rows so has_more is known without a count. The
        // extra row is truncated away before returning.
        let fetch = limit + 1;
        let data_query = if cursor_walk {
            format!(
                "SELECT * FROM {}{}{} LIMIT {}",
                self.table_name, where_clause, order_clause, fetch
            )
        } else {
            format!(
                "SELECT * FROM {}{}{} LIMIT {} OFFSET {}",
                self.table_name,
                where_clause,
                order_clause,
                fetch,
                pagination.offset()
            )
        };

        let mut pagination_info = PaginationInfo::new(pagination.page, pagination.per_page, total);
        pagination_info.count_mode = count_mode.to_string();

        // One fetch path for both modes: untyped rows, decoded per-entity
        // through FromRow exactly as query_as would (decimals keep their
        // scale), with the boundary values read for the cursors off the same
        // rows — no second query, no JSON round-trip.
        let mut rows_query = sqlx::query(&data_query);
        for param in &filter_params {
            rows_query = rows_query.bind(param);
        }
        let rows: Vec<PgRow> =
            crate::company_scope::fetch_all_rows_scoped(&self.pool, rows_query).await?;
        let has_more = rows.len() as i64 > limit;
        let mut page: Vec<PgRow> = rows.into_iter().take(limit as usize).collect();
        if backwards {
            page.reverse();
        }
        let data: anyhow::Result<Vec<T>> = page
            .iter()
            .map(|row| T::from_row(row).map_err(|e| anyhow::anyhow!("decode row: {e}")))
            .collect();
        let data = data?;

        // Cursors from the boundary rows' own values, whenever the order is
        // deterministic. A NULL in a sort column cannot key a position, so
        // that side's cursor is omitted.
        let deterministic = !boundary_sorts.is_empty();
        let next_cursor = if (has_more || backwards) && deterministic {
            page.last().and_then(|r| {
                let casts = if boundary_casts.is_empty() {
                    // Page mode never resolved casts; the encode does not
                    // need them, only the walk does.
                    &boundary_casts
                } else {
                    &boundary_casts
                };
                self.row_cursor(r, &boundary_sorts, casts)
            })
        } else {
            None
        };
        let prev_cursor = if !page.is_empty() && deterministic {
            page.first().and_then(|r| self.row_cursor(r, &boundary_sorts, &boundary_casts))
        } else {
            None
        };

        pagination_info.has_more = Some(has_more);
        pagination_info.next_cursor = next_cursor;
        pagination_info.prev_cursor = prev_cursor;

        Ok(PaginatedResult {
            data,
            pagination: pagination_info,
        })
    }

    /// The planner's row estimate for a filtered read, from EXPLAIN. The
    /// exact figure stays the separate count endpoint; this exists because
    /// counting a hot filtered set is a scan of all of it.
    /// The exact number of rows the filter matches: a scan of the whole filtered set.
    async fn count_filtered_rows(
        &self,
        where_clause: &str,
        filter_params: &[String],
    ) -> anyhow::Result<u64> {
        let count_query = format!("SELECT COUNT(*) FROM {}{}", self.table_name, where_clause);
        let mut count_query_builder = sqlx::query_scalar::<_, i64>(&count_query);
        for param in filter_params {
            count_query_builder = count_query_builder.bind(param);
        }
        Ok(crate::company_scope::fetch_one_scalar_scoped(&self.pool, count_query_builder).await?
            as u64)
    }

    async fn estimate_filtered_rows(
        &self,
        where_clause: &str,
        filter_params: &[String],
    ) -> anyhow::Result<u64> {
        let explain = format!("EXPLAIN (FORMAT JSON) SELECT 1 FROM {}{}", self.table_name, where_clause);
        let mut builder = sqlx::query_scalar::<_, serde_json::Value>(&explain);
        for param in filter_params {
            builder = builder.bind(param);
        }
        let plan: serde_json::Value =
            crate::company_scope::fetch_one_scalar_scoped(&self.pool, builder).await?;
        let rows = plan
            .as_array()
            .and_then(|a| a.first())
            .and_then(|top| top.get("Plan"))
            .and_then(|p| p.get("Plan Rows"))
            .and_then(|r| r.as_i64())
            .unwrap_or(0);
        Ok(rows.max(0) as u64)
    }

    /// The SQL cast suffix for each sort column, from the table's real
    /// columns (never a cached hint list — the aggregate lesson). The
    /// placeholder is cast to the column's type so the bind compares
    /// against the column without coercing the column itself.
    async fn sort_column_casts(
        &self,
        sorts: &[(String, FilterSortDirection)],
    ) -> anyhow::Result<Vec<Option<String>>> {
        let (schema, table) = match self.table_name.rsplit_once('.') {
            Some((s, t)) => (s.to_string(), t.to_string()),
            None => ("public".to_string(), self.table_name.clone()),
        };
        let mut casts: Vec<Option<String>> = Vec::with_capacity(sorts.len());
        for (field, _) in sorts {
            let row: Option<(String, String, String)> = sqlx::query_as(
                "SELECT data_type, coalesce(udt_name, ''), coalesce(udt_schema, '') \
                   FROM information_schema.columns \
                  WHERE table_schema = $1 AND table_name = $2 AND column_name = $3",
            )
            .bind(&schema)
            .bind(&table)
            .bind(field)
            .fetch_optional(&self.pool)
            .await?;
            let cast = row
                .map(|(data_type, udt, udt_schema)| cast_suffix(&data_type, &udt, &udt_schema))
                .flatten();
            casts.push(cast);
        }
        Ok(casts)
    }

    /// One boundary row's cursor: its values for the sort columns (typed
    /// exactly as sqlx decodes them, so a decimal keeps its scale) plus its
    /// id. NULL in any sort column yields None: a null has no position.
    fn row_cursor(
        &self,
        row: &PgRow,
        sorts: &[(String, FilterSortDirection)],
        casts: &[Option<String>],
    ) -> Option<String> {
        let id: uuid::Uuid = row.try_get("id").ok()?;
        let mut values: Vec<serde_json::Value> = Vec::with_capacity(sorts.len());
        for (i, (field, _)) in sorts.iter().enumerate() {
            let field = field.as_str();
            let mut data_type = casts.get(i).and_then(|c| c.as_deref()).unwrap_or("");
            // The ORM appends the id tiebreaker itself and page mode never
            // resolves casts — without this, the uuid id decodes as text,
            // the read fails, and the whole cursor silently vanishes.
            if field == "id" && data_type.is_empty() {
                data_type = "uuid";
            }
            let text: Option<String> = match data_type {
                "numeric" => row
                    .try_get::<Option<sqlx::types::Decimal>, _>(field)
                    .ok()?
                    .map(|d| d.to_string()),
                "uuid" => row
                    .try_get::<Option<uuid::Uuid>, _>(field)
                    .ok()?
                    .map(|u| u.to_string()),
                "timestamptz" => row
                    .try_get::<Option<chrono::DateTime<chrono::Utc>>, _>(field)
                    .ok()?
                    .map(|t| t.to_rfc3339()),
                "integer" | "smallint" => row
                    .try_get::<Option<i32>, _>(field)
                    .ok()?
                    .map(|n| n.to_string()),
                "bigint" => row
                    .try_get::<Option<i64>, _>(field)
                    .ok()?
                    .map(|n| n.to_string()),
                "boolean" => row
                    .try_get::<Option<bool>, _>(field)
                    .ok()?
                    .map(|b| b.to_string()),
                "date" => row
                    .try_get::<Option<chrono::NaiveDate>, _>(field)
                    .ok()?
                    .map(|d| d.to_string()),
                "timestamp" => row
                    .try_get::<Option<chrono::NaiveDateTime>, _>(field)
                    .ok()?
                    .map(|t| t.format("%Y-%m-%d %H:%M:%S%.6f").to_string()),
                // A text column: no cast, decoded as text.
                "" => row.try_get::<Option<String>, _>(field).ok()?,
                // A column of an enum type, whose cast names the type: Postgres sends the
                // label as text, but a checked decode as `String` refuses a type that is not
                // `text`, which dropped the cursor for every page of a list sorted by an enum
                // (the page said more rows followed and handed out no way to continue).
                _ => row.try_get_unchecked::<Option<String>, _>(field).ok()?,
            };
            values.push(serde_json::Value::String(text?));
        }
        crate::filter::cursor::encode_cursor(sorts, &values, &id.to_string()).ok()
    }
}

/// The placeholder cast for a column type, or None when a bare text bind
/// compares correctly.
/// The SQL of one guarded row update — see [`PostgresRepository::update_guarded`].
pub(crate) enum UpdateStatement {
    /// Nothing to write: re-read the row under the guard. Binds `$1` = id.
    Unchanged(String),
    /// The UPDATE. Binds `$1` = the row as JSON, `$2` = id.
    Update { query: String, json: String },
}

/// Build the guarded update of `entity` in `table`, setting only `only` columns when given.
pub(crate) fn update_statement<T: Serialize>(
    table: &str,
    entity: &T,
    only: Option<&[String]>,
    guard: &str,
) -> anyhow::Result<UpdateStatement> {
    let json_obj = match serde_json::to_value(entity)? {
        Value::Object(obj) => obj,
        _ => return Err(anyhow::anyhow!("Entity must serialize to a JSON object")),
    };

    let update_columns: Vec<&String> = json_obj
        .keys()
        .filter(|k| *k != "id")
        .filter(|k| only.is_none_or(|cols| cols.iter().any(|c| c == *k)))
        .collect();

    if update_columns.is_empty() {
        return Ok(UpdateStatement::Unchanged(format!(
            "SELECT t.* FROM {table} AS t WHERE t.id = $1::uuid{guard}"
        )));
    }

    let column_names = update_columns
        .iter()
        .map(|k| quote_ident(k))
        .collect::<Vec<_>>()
        .join(", ");
    let json = serde_json::to_string(&json_obj)?;

    // jsonb_populate_record in a CTE gives every value its column's real type (enums,
    // booleans, numerics) without per-column casts.
    let query = format!(
        r#"
            WITH new_row AS (
                SELECT (jsonb_populate_record(NULL::{table}, $1::jsonb)).*
            )
            UPDATE {table} AS t
            SET ({columns}) = (SELECT {columns} FROM new_row)
            WHERE t.id = $2::uuid{guard}
            RETURNING t.*
            "#,
        table = table,
        columns = column_names,
        guard = guard,
    );
    Ok(UpdateStatement::Update { query, json })
}

/// The SQL of one insert: the query, and the JSON payload it binds as `$1` (`None` for the
/// `DEFAULT VALUES` form, which binds nothing).
pub(crate) fn insert_statement<T: Serialize>(
    table: &str,
    entity: &T,
) -> anyhow::Result<(String, Option<String>)> {
    let json_obj = match serde_json::to_value(entity)? {
        Value::Object(obj) => obj,
        _ => return Err(anyhow::anyhow!("Entity must serialize to a JSON object")),
    };

    // Name only the columns the payload actually carries.
    //
    // `SELECT (jsonb_populate_record(...)).*` emits EVERY column of the row type, so a column
    // the entity does not know about arrived as an explicit NULL — and an explicit NULL is not
    // an absent value: it overrides the column DEFAULT. That is invisible until a table's
    // correctness depends on a default, which is exactly what composition-installed tenancy
    // does (a scoped table defaults `org_unit_id` from the acting unit), so generic creates
    // over such a table wrote NULL and were refused by the write-path guard.
    //
    // Listing the payload's own keys leaves every other column unmentioned, so its default
    // applies. A key that is present with a JSON null is still written as NULL, which is
    // right: the caller said so. This mirrors the update path, which has always built its
    // column list from these same keys.
    let insert_columns: Vec<String> = json_obj.keys().map(|k| quote_ident(k)).collect();

    if insert_columns.is_empty() {
        // Nothing supplied at all: let every column take its default rather than emitting
        // `INSERT INTO t () SELECT`, which is not valid SQL.
        return Ok((format!("INSERT INTO {table} DEFAULT VALUES RETURNING *"), None));
    }
    let columns = insert_columns.join(", ");
    let query = format!(
        r#"
            INSERT INTO {table} ({columns})
            SELECT {columns} FROM jsonb_populate_record(NULL::{table}, $1::jsonb)
            RETURNING *
            "#,
    );
    Ok((query, Some(serde_json::to_string(&json_obj)?)))
}

/// The serialized fields whose value differs between `before` and `after` (never `id`), in
/// `after`'s field order — the columns an update that started from `before` actually changes.
/// A field `after` no longer serializes is not a change: an update cannot unset a column by
/// leaving it out.
pub fn changed_columns<T: Serialize>(before: &T, after: &T) -> anyhow::Result<Vec<String>> {
    let (Value::Object(before), Value::Object(after)) =
        (serde_json::to_value(before)?, serde_json::to_value(after)?)
    else {
        return Err(anyhow::anyhow!("Entity must serialize to a JSON object"));
    };
    Ok(after
        .into_iter()
        .filter(|(k, v)| k != "id" && before.get(k) != Some(v))
        .map(|(k, _)| k)
        .collect())
}

/// The primary-key column every generated table carries, and the final key of
/// every paged read's order.
const KEY_COLUMN: &str = "id";

/// Make a requested sort a TOTAL order.
///
/// One rule for every paged read (list, filtered list, search, trash, keyset
/// walk): the order is the caller's sorts followed by the primary key `id`
/// ascending, unless `id` is already one of the sorts — it is unique, so
/// anything after it could never be consulted. With no sort at all the order
/// is `id` ascending.
///
/// Without a final unique key Postgres is free to return rows that tie (and,
/// with no `ORDER BY` at all, every row) in physical order, which changes
/// after an update or a vacuum — so offset paging could repeat a row on one
/// page and skip another between two fetches. `id` is the primary key, so the
/// order can ride its index with `LIMIT` when nothing else is sorted.
pub(crate) fn total_order(
    mut sorts: Vec<(String, FilterSortDirection)>,
) -> Vec<(String, FilterSortDirection)> {
    if !sorts.iter().any(|(f, _)| f == KEY_COLUMN) {
        sorts.push((KEY_COLUMN.into(), FilterSortDirection::Asc));
    }
    sorts
}

/// ` ORDER BY ...` for a total order. `backwards` inverts every direction,
/// which is how a `before=` walk reads the page that precedes a cursor; the
/// caller reverses the rows afterwards.
pub(crate) fn order_by_sql(sorts: &[(String, FilterSortDirection)], backwards: bool) -> String {
    let parts: Vec<String> = sorts
        .iter()
        .map(|(f, d)| {
            let dir = if (*d == FilterSortDirection::Desc) != backwards { "DESC" } else { "ASC" };
            format!("{} {}", f, dir)
        })
        .collect();
    format!(" ORDER BY {}", parts.join(", "))
}

/// AND the keyset predicate onto a built WHERE clause. The existing clause is parenthesised
/// as a whole first: a lone group may carry a client `OR`, which would otherwise bind looser
/// than the keyset and let rows from outside the page window through.
fn and_keyset_predicate(where_clause: &str, keyset_sql: &str) -> String {
    match where_clause.trim_start().strip_prefix("WHERE ") {
        None => format!(" WHERE {}", keyset_sql),
        Some(existing) => format!(" WHERE ({}) AND ({})", existing, keyset_sql),
    }
}

fn cast_suffix(data_type: &str, udt_name: &str, udt_schema: &str) -> Option<String> {
    match data_type {
        "uuid" => Some("uuid".into()),
        "numeric" => Some("numeric".into()),
        "integer" => Some("integer".into()),
        "smallint" => Some("smallint".into()),
        "bigint" => Some("bigint".into()),
        "boolean" => Some("boolean".into()),
        "date" => Some("date".into()),
        "timestamp with time zone" => Some("timestamptz".into()),
        "timestamp without time zone" => Some("timestamp".into()),
        // An enum: bind text, cast to the enum's own name so the comparison
        // runs in the enum's ordering — schema-qualified, so it resolves on a
        // tenant whose search path does not hold the module's schema.
        "USER-DEFINED" if !udt_name.is_empty() => Some(match udt_schema {
            "" | "public" | "pg_catalog" => udt_name.to_string(),
            schema => format!("{schema}.{udt_name}"),
        }),
        _ => None,
    }
}

/// The cast a filter placeholder needs to compare against a column, from the column's catalog
/// type: `format_type(atttypid, NULL)` and `pg_type.typtype`. None when a bare text bind already
/// compares correctly (text-like columns) or when no single-value cast fits (arrays, json,
/// composite and domain types keep today's text bind).
fn filter_cast_for(type_name: &str, typtype: &str) -> Option<String> {
    match typtype {
        // An enum: cast to the enum itself so the comparison runs in the enum's ordering. The
        // name comes from `format_type`, schema-qualified when the type is not on the search path.
        "e" => Some(type_name.to_string()),
        "b" => match type_name {
            "uuid" | "boolean" | "smallint" | "integer" | "bigint" | "numeric" | "real"
            | "double precision" | "date" | "interval" | "inet" | "cidr" | "macaddr" => {
                Some(type_name.to_string())
            }
            "time without time zone" => Some("time".into()),
            "time with time zone" => Some("timetz".into()),
            "timestamp with time zone" => Some("timestamptz".into()),
            "timestamp without time zone" => Some("timestamp".into()),
            _ => None,
        },
        _ => None,
    }
}

/// Every column of `qualified_table` whose filter placeholder needs a cast, with that cast.
///
/// Read from `pg_catalog` rather than `information_schema`: it is a direct lookup on the
/// relation, it says whether a user-defined type is an enum (a domain or composite must not be
/// treated as one), and `to_regclass` resolves the name exactly as the query will. Runs through
/// the company-scoped helper so it uses the request's own connection when one is held.
async fn catalog_filter_casts(
    pool: &PgPool,
    qualified_table: &str,
) -> anyhow::Result<HashMap<String, String>> {
    let q = sqlx::query_as::<Postgres, (String, String, String)>(
        "SELECT a.attname::text, format_type(a.atttypid, NULL), t.typtype::text
           FROM pg_catalog.pg_attribute a
           JOIN pg_catalog.pg_type t ON t.oid = a.atttypid
          WHERE a.attrelid = to_regclass($1) AND a.attnum > 0 AND NOT a.attisdropped",
    )
    .bind(qualified_table.to_string());
    let rows = crate::company_scope::fetch_all_scoped(pool, q).await?;
    Ok(rows
        .into_iter()
        .filter_map(|(column, type_name, typtype)| {
            filter_cast_for(&type_name, &typtype).map(|cast| (column, cast))
        })
        .collect())
}

/// The generated hints, with the catalog's casts filling every column they do not name. A hint
/// keeps deciding its own column, so nothing that compares correctly today changes — except that
/// a hint naming an enum without its schema (`task_status`) gives way to the catalog's qualified
/// name for the same type (`lifecycle.task_status`), which resolves whatever the search path.
fn merge_filter_casts(
    hints: &HashMap<String, String>,
    mut catalog: HashMap<String, String>,
) -> HashMap<String, String> {
    for (column, cast) in hints {
        let qualified_same_type = catalog
            .get(column)
            .and_then(|c| c.rsplit_once('.'))
            .is_some_and(|(_, name)| name.trim_matches('"') == cast.as_str());
        if !qualified_same_type {
            catalog.insert(column.clone(), cast.clone());
        }
    }
    catalog
}

/// Turn "column ... does not exist" into a sentence that names the cause.
///
/// The insert names the columns the entity serializes. A field that is serialized but is not a
/// column of the table used to vanish quietly, because selecting every column of the row type threw
/// unknown keys away; now it fails, and the bare Postgres error does not say why. Serialized field
/// and table column are meant to be the same set — the update path has always assumed it — so this
/// points at the mismatch rather than leaving someone to guess.
pub(crate) fn explain_unknown_column(error: sqlx::Error, table: &str) -> anyhow::Error {
    let text = error.to_string();
    if text.contains("does not exist") && text.contains("column") {
        return anyhow::Error::new(error).context(format!(
            "insert into {table} named a column that does not exist: the entity serializes a field \
             with no matching column. Every serialized field must be a column of the table (rename \
             it, map it with #[serde(rename)], or skip it with #[serde(skip)])"
        ));
    }
    anyhow::Error::new(error)
}

/// Quote a column name as a SQL identifier.
///
/// Column names here come from serializing the caller's entity, so they are Rust field names in
/// practice — but they are interpolated into DDL/DML, where Postgres has no bind parameter for an
/// identifier. Doubling an embedded quote is the identifier escape, so a name can never end the
/// quoted section early.
fn quote_ident(name: &str) -> String {
    format!("\"{}\"", name.replace('"', "\"\""))
}

#[async_trait]
impl<T> DatabaseOperations<T> for PostgresRepository<T>
where
    T: for<'a> FromRow<'a, PgRow> + Send + Sync + Unpin + Serialize,
{
    async fn create(&self, entity: &T) -> anyhow::Result<T> {
        let (query, json) = insert_statement(&self.table_name, entity)?;
        // The DEFAULT VALUES form takes no bind; every other form binds the payload.
        let statement = match json {
            Some(json) => sqlx::query_as::<_, T>(&query).bind(json),
            None => sqlx::query_as::<_, T>(&query),
        };
        let result = crate::company_scope::fetch_one_scoped(&self.pool, statement)
            .await
            .map_err(|e| explain_unknown_column(e, &self.table_name))?;

        Ok(result)
    }

    async fn find_by_id(&self, id: &str) -> anyhow::Result<Option<T>> {
        // Cast text to UUID for PostgreSQL UUID columns
        let query = format!("SELECT * FROM {} WHERE id = $1::uuid", self.table_name);
        let result = crate::company_scope::fetch_optional_scoped(
            &self.pool,
            sqlx::query_as::<Postgres, T>(&query).bind(id),
        )
        .await?;
        Ok(result)
    }

    async fn find_all(&self) -> anyhow::Result<Vec<T>> {
        let query = format!("SELECT * FROM {}", self.table_name);
        let results = crate::company_scope::fetch_all_scoped(
            &self.pool,
            sqlx::query_as::<Postgres, T>(&query),
        )
        .await?;
        Ok(results)
    }

    async fn update(&self, id: &str, entity: &T) -> anyhow::Result<Option<T>> {
        self.update_guarded(id, entity, None, "").await
    }

    async fn delete(&self, id: &str) -> anyhow::Result<bool> {
        let query = format!("DELETE FROM {} WHERE id = $1::uuid", self.table_name);
        let result = crate::company_scope::execute_scoped(
            &self.pool,
            sqlx::query(&query).bind(id),
        )
        .await?;
        Ok(result.rows_affected() > 0)
    }

    async fn count(&self) -> anyhow::Result<u64> {
        let query = format!("SELECT COUNT(*) FROM {}", self.table_name);
        let count = crate::company_scope::fetch_one_scalar_scoped(
            &self.pool,
            sqlx::query_scalar::<_, i64>(&query),
        )
        .await? as u64;
        Ok(count)
    }

    async fn exists(&self, id: &str) -> anyhow::Result<bool> {
        let query = format!("SELECT 1 FROM {} WHERE id = $1::uuid LIMIT 1", self.table_name);
        let result = crate::company_scope::fetch_optional_scalar_scoped(
            &self.pool,
            sqlx::query_scalar::<_, i32>(&query).bind(id),
        )
        .await?;
        Ok(result.is_some())
    }

    async fn execute_query(&self, query: &str) -> anyhow::Result<u64> {
        let result = crate::company_scope::execute_scoped(
            &self.pool,
            sqlx::query(query),
        )
        .await?;
        Ok(result.rows_affected())
    }
}

// ─── Aggregation ──────────────────────────────────────────────────────────────

/// Which reduction to apply to a column.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
pub enum AggregateFn {
    Sum,
    Avg,
    Min,
    Max,
}

impl AggregateFn {
    fn sql(self) -> &'static str {
        match self {
            AggregateFn::Sum => "SUM",
            AggregateFn::Avg => "AVG",
            AggregateFn::Min => "MIN",
            AggregateFn::Max => "MAX",
        }
    }

    fn label(self) -> &'static str {
        match self {
            AggregateFn::Sum => "sum",
            AggregateFn::Avg => "avg",
            AggregateFn::Min => "min",
            AggregateFn::Max => "max",
        }
    }

    /// `SUM`/`AVG` on a text column is a type error, not a zero. `MIN`/`MAX`
    /// order any comparable type, so they carry no such restriction.
    fn requires_numeric(self) -> bool {
        matches!(self, AggregateFn::Sum | AggregateFn::Avg)
    }
}

/// What to group by and what to reduce — the parsed form of the query string.
#[derive(Debug, Clone, Default)]
pub struct AggregateSpec {
    /// Column whose distinct values become groups. `None` asks for one total.
    pub group_by: Option<String>,
    /// `(function, column)` pairs, in the order the caller asked for them.
    pub reductions: Vec<(AggregateFn, String)>,
    /// Most groups to return before reporting the answer as truncated.
    pub group_limit: usize,
    /// Carry a label per group: the column on the group column's RELATED
    /// table (resolved through the entity's relation metadata) to show
    /// instead of a bare uuid key. `None` = keys stay as they are.
    pub label_field: Option<String>,
    /// The resolved relation behind the group column — `(target table, the
    /// BASE table's FK column, snake)`, filled by the generic layer from
    /// the entity's `relations()` metadata. Callers never set this.
    pub label_relation: Option<(String, String)>,
}

/// The default ceiling on distinct groups.
///
/// A `group_by` on a uuid or a timestamp yields one group per row, which is a
/// table scan wearing a chart's clothes. Rather than refusing those columns —
/// a list that would be wrong for some schema sooner or later — the answer is
/// capped and the cap is *reported*, so a caller can tell a complete picture
/// from a partial one instead of quietly drawing the wrong one.
pub const DEFAULT_GROUP_LIMIT: usize = 200;

/// One group's numbers. `key` is the group's value; `None` is a real answer —
/// the rows whose group column is null — and is distinct from "no rows".
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct AggregateGroup {
    pub key: Option<String>,
    /// The group's display label (the related row's label column), when the
    /// caller asked for one and the group column is a relation FK.
    pub label: Option<String>,
    pub count: u64,
    /// Reduction results keyed `"sum:amount"`, carried as strings.
    ///
    /// Postgres `numeric` holds more precision than an IEEE double, and money
    /// columns are exactly where that bites: a tenant large enough for the
    /// total to matter is a tenant large enough to round it. The string is the
    /// exact value Postgres computed; the caller decides how to parse it.
    /// `None` is SQL NULL — no rows contributed — which is not zero.
    pub values: HashMap<String, Option<String>>,
}

/// Groups plus the overall total, computed together.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct AggregateResult {
    pub groups: Vec<AggregateGroup>,
    pub total: AggregateGroup,
    /// True when more distinct groups exist than `group_limit` allowed.
    pub truncated: bool,
}

/// A column name rejected by the allow-list, or a reduction that its type
/// cannot answer.
#[derive(Debug, Clone)]
pub struct AggregateFieldError(pub String);

impl std::fmt::Display for AggregateFieldError {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.write_str(&self.0)
    }
}

impl std::error::Error for AggregateFieldError {}

/// True for the Postgres types `SUM`/`AVG` accept.
fn is_numeric_pg_type(pg_type: &str) -> bool {
    let t = pg_type.trim().to_ascii_lowercase();
    let t = t.split('(').next().unwrap_or(&t).trim();
    matches!(
        t,
        "numeric" | "decimal" | "money"
            | "smallint" | "int2" | "integer" | "int" | "int4" | "bigint" | "int8"
            | "real" | "float4" | "double precision" | "float8"
            | "smallserial" | "serial" | "bigserial"
    )
}

/// Read a table's real columns and their types from the catalog.
///
/// `EntityRepoMeta::column_types()` looks like the natural allow-list and is
/// not one: it carries only the columns the filter parser must CAST — uuids and
/// enums — so every numeric column is absent from it, which is exactly the set
/// `sum` and `avg` exist for. The catalog is the only complete and current
/// answer, and it cannot drift from the table the query will actually run
/// against.
async fn catalog_columns(
    pool: &PgPool,
    qualified_table: &str,
) -> anyhow::Result<HashMap<String, String>> {
    let (schema, table) = match qualified_table.split_once('.') {
        Some((s, t)) => (s.to_string(), t.to_string()),
        None => ("public".to_string(), qualified_table.to_string()),
    };
    let q = sqlx::query_as::<Postgres, (String, String)>(
        "SELECT column_name, data_type FROM information_schema.columns
          WHERE table_schema = $1 AND table_name = $2",
    )
    .bind(schema)
    .bind(table);
    let rows = crate::company_scope::fetch_all_scoped(pool, q).await?;
    Ok(rows.into_iter().collect())
}

/// Resolve a caller-supplied column name against the entity's real columns.
///
/// This is the whole defence for the aggregate path. Unlike a filter *value*,
/// which is bound as a parameter, a `group_by` or `sum` column is spliced into
/// the SQL as an identifier — binding cannot protect it. So the name is never
/// escaped or quoted into safety; it is *replaced* by the matching key already
/// present in the entity's declared column map, and a name with no match is
/// refused. Nothing a caller types can reach the query text.
fn resolve_column<'a>(
    name: &str,
    column_types: &'a HashMap<String, String>,
) -> Result<(&'a str, &'a str), AggregateFieldError> {
    column_types
        .get_key_value(name)
        .map(|(k, v)| (k.as_str(), v.as_str()))
        .ok_or_else(|| {
            AggregateFieldError(format!("unknown field `{name}` — not a column of this entity"))
        })
}

impl<T: for<'a> FromRow<'a, PgRow> + Send + Unpin> PostgresRepository<T> {
    /// Group and reduce rows in one statement, under the same filters, the same
    /// soft-delete convention and the same tenancy fence as the list endpoint.
    ///
    /// Groups and the overall total come back from a single `GROUPING SETS`
    /// query, which is what lets a caller draw a chart and its headline from
    /// one reply, and what keeps the two numbers consistent — a separate total
    /// query could observe a different set of rows.
    pub async fn aggregate_with_filters(
        &self,
        spec: &AggregateSpec,
        filters: &HashMap<String, String>,
        column_types: &HashMap<String, String>,
        search_fields: &[&str],
    ) -> anyhow::Result<AggregateResult> {
        self.aggregate_with_filters_and_base(spec, None, filters, column_types, search_fields)
            .await
    }

    /// [`Self::aggregate_with_filters`] under a repository-authored base condition — the same
    /// contract as [`Self::list_with_filters_and_base`], so an aggregate reduces exactly the
    /// rows the matching list returns.
    pub(crate) async fn aggregate_with_filters_and_base(
        &self,
        spec: &AggregateSpec,
        base_condition: Option<&str>,
        filters: &HashMap<String, String>,
        column_types: &HashMap<String, String>,
        search_fields: &[&str],
    ) -> anyhow::Result<AggregateResult> {
        match self
            .aggregate_unattributed(spec, base_condition, filters, column_types, search_fields)
            .await
        {
            Ok(result) => Ok(result),
            Err(e) => Err(self.name_refused_parameter(e, filters).await),
        }
    }

    /// Name the query parameter behind a failed statement (see
    /// [`crate::filter::attribute_query_error`]). An operator the column's type lacks is
    /// ambiguous when several filters use pattern operators, so that one case reads the table's
    /// column types — through the pool, not the request's connection, which the failed
    /// statement may have left in an aborted transaction.
    async fn name_refused_parameter(
        &self,
        err: anyhow::Error,
        filters: &HashMap<String, String>,
    ) -> anyhow::Error {
        let err = crate::filter::attribute_query_error(err, filters);
        if crate::filter::undefined_operator(&err).is_none() {
            return err;
        }
        let (schema, table) = match self.table_name.split_once('.') {
            Some((s, t)) => (s.to_string(), t.to_string()),
            None => ("public".to_string(), self.table_name.clone()),
        };
        let types: Vec<(String, String)> = sqlx::query_as(
            "SELECT column_name, data_type FROM information_schema.columns \
              WHERE table_schema = $1 AND table_name = $2",
        )
        .bind(schema)
        .bind(table)
        .fetch_all(&self.pool)
        .await
        .unwrap_or_default();
        let types: HashMap<String, String> = types.into_iter().collect();
        crate::filter::attribute_operator_error(err, filters, &types)
    }

    /// [`Self::aggregate_with_filters_and_base`] before a database refusal is matched to the
    /// query parameter behind it.
    async fn aggregate_unattributed(
        &self,
        spec: &AggregateSpec,
        base_condition: Option<&str>,
        filters: &HashMap<String, String>,
        column_types: &HashMap<String, String>,
        search_fields: &[&str],
    ) -> anyhow::Result<AggregateResult> {
        let mut query_filter = self.parse_typed_filters(filters, column_types, None).await?;
        if let Some(cond) = base_condition {
            query_filter.add_base_condition(cond);
        }
        if !search_fields.is_empty() {
            query_filter.search_fields = search_fields.iter().map(|s| s.to_string()).collect();
        }
        // Grouping replaces row output entirely: paging and ordering describe a
        // page of rows, and there are none.
        query_filter.limit = None;
        query_filter.offset = None;
        let (where_clause, filter_params) = query_filter.build_where_clause();

        // Every identifier below comes from the catalog, never from the caller.
        let columns = catalog_columns(&self.pool, &self.table_name).await?;
        let mut selects: Vec<String> = Vec::new();
        let mut value_keys: Vec<String> = Vec::new();
        for (func, field) in &spec.reductions {
            let (column, pg_type) = resolve_column(field, &columns)
                .map_err(|_| crate::filter::QueryParamError::unknown_column(func.label(), field))?;
            if func.requires_numeric() && !is_numeric_pg_type(pg_type) {
                return Err(crate::filter::QueryParamError::new(
                    func.label(),
                    crate::filter::QueryParamReason::NotNumeric,
                    format!(
                        "`{label}`: `{field}` is not a numeric field, and {label} needs one",
                        label = func.label()
                    ),
                )
                .into());
            }
            let key = format!("{}:{}", func.label(), column);
            // Cast to text in SQL so the exact value Postgres computed is what
            // crosses the wire — see `AggregateGroup::values`.
            selects.push(format!("{}({})::text AS \"{}\"", func.sql(), column, key));
            value_keys.push(key);
        }

        let group_limit = if spec.group_limit == 0 { DEFAULT_GROUP_LIMIT } else { spec.group_limit };
        let reductions = if selects.is_empty() { String::new() } else { format!(", {}", selects.join(", ")) };

        // The group's label: when the caller names one (group_label) and the
        // group column is a relation FK of THIS entity, LEFT JOIN the
        // related table and carry its label column beside the key — the
        // aggregate's equivalent of `?include=`, which has no row to
        // hydrate otherwise.
        let mut label_select = String::new();
        let mut label_join = String::new();
        if let (Some(field), Some(label_field), Some((rel_table, base_fk))) = (
            &spec.group_by,
            &spec.label_field,
            &spec.label_relation,
        ) {
            let _ = field;
            let qualified = qualify_relation_table(&self.table_name, rel_table);
            let rel_columns = catalog_columns(&self.pool, &qualified).await?;
            let (label_col, _) = resolve_column(label_field, &rel_columns).map_err(|_| {
                crate::filter::QueryParamError::new(
                    "group_label",
                    crate::filter::QueryParamReason::UnknownColumn,
                    format!("`group_label`: `{label_field}` is not a field of the records the groups point at"),
                )
            })?;
            label_select = format!(", (label_rel.{label_col})::text AS __group_label");
            label_join = format!(
                " LEFT JOIN {qualified} AS label_rel ON label_rel.id IS NOT DISTINCT FROM {base_fk}"
            );
        }

        let sql = match &spec.group_by {
            Some(field) => {
                let (column, _) = resolve_column(field, &columns).map_err(|_| {
                    crate::filter::QueryParamError::unknown_column("group_by", field)
                })?;
                format!(
                    "SELECT GROUPING({column}) AS __is_total, ({column})::text AS __group_key{label_select}, \
                     COUNT(*) AS __count{reductions} \
                     FROM {table}{label_join}{where_clause} \
                     GROUP BY GROUPING SETS (({column}), ()) \
                     ORDER BY __is_total DESC, __count DESC, __group_key \
                     LIMIT {limit}",
                    column = column,
                    label_select = label_select,
                    label_join = label_join,
                    reductions = reductions,
                    table = self.table_name,
                    where_clause = where_clause,
                    // One total row, the groups themselves, and one more to
                    // detect that a further group existed.
                    limit = group_limit + 2,
                )
            }
            None => format!(
                "SELECT 1 AS __is_total, NULL::text AS __group_key, COUNT(*) AS __count{reductions} \
                 FROM {table}{where_clause}",
                reductions = reductions,
                table = self.table_name,
                where_clause = where_clause,
            ),
        };

        let mut builder = sqlx::query(&sql);
        for param in &filter_params {
            builder = builder.bind(param);
        }
        let rows = crate::company_scope::fetch_all_rows_scoped(&self.pool, builder).await?;

        let read_group = |row: &PgRow| -> AggregateGroup {
            use sqlx::Row as _;
            let mut values = HashMap::with_capacity(value_keys.len());
            for key in &value_keys {
                values.insert(key.clone(), row.try_get::<Option<String>, _>(key.as_str()).ok().flatten());
            }
            AggregateGroup {
                key: row.try_get::<Option<String>, _>("__group_key").ok().flatten(),
                label: row.try_get::<Option<String>, _>("__group_label").ok().flatten(),
                count: row.try_get::<i64, _>("__count").unwrap_or(0).max(0) as u64,
                values,
            }
        };

        use sqlx::Row as _;
        let mut total: Option<AggregateGroup> = None;
        let mut groups: Vec<AggregateGroup> = Vec::new();
        for row in &rows {
            let is_total = row.try_get::<i32, _>("__is_total").unwrap_or(0) == 1;
            if is_total {
                // Ordered first, so it survives the cap.
                total = Some(read_group(row));
            } else {
                groups.push(read_group(row));
            }
        }

        let truncated = groups.len() > group_limit;
        groups.truncate(group_limit);

        // No rows at all means no total row either: an empty result is a real
        // answer of zero, not a missing one.
        let total = total.unwrap_or_else(|| AggregateGroup {
            key: None,
            label: None,
            count: 0,
            values: value_keys.iter().map(|k| (k.clone(), None)).collect(),
        });

        Ok(AggregateResult { groups, total, truncated })
    }
}

#[cfg(test)]
mod aggregate_field_tests {
    use super::*;

    fn columns() -> HashMap<String, String> {
        [
            ("status", "text"),
            ("total", "numeric"),
            ("qty", "integer"),
            ("notes", "text"),
        ]
        .iter()
        .map(|(k, v)| (k.to_string(), v.to_string()))
        .collect()
    }

    #[test]
    fn resolves_only_declared_columns() {
        let cols = columns();
        assert_eq!(resolve_column("total", &cols).unwrap().0, "total");
        assert!(resolve_column("password_hash", &cols).is_err());
    }

    /// The allow-list is the entire defence, because a group/sum column is
    /// spliced into SQL as an identifier and cannot be bound as a parameter.
    #[test]
    fn rejects_injection_attempts_rather_than_escaping_them() {
        let cols = columns();
        for probe in [
            "total) FROM selling.sales_orders; DROP TABLE users --",
            "status\"",
            "1=1",
            "total, (SELECT password FROM users)",
            "",
        ] {
            assert!(
                resolve_column(probe, &cols).is_err(),
                "`{probe}` must be refused, never escaped into the query"
            );
        }
    }

    /// The returned name is the map's own key, not the caller's string, so no
    /// caller-controlled bytes can reach the SQL even on a match.
    #[test]
    fn returns_the_declared_key_not_the_callers_string() {
        let cols = columns();
        let (name, _) = resolve_column("total", &cols).unwrap();
        assert!(std::ptr::eq(name, cols.get_key_value("total").unwrap().0.as_str()));
    }

    #[test]
    fn sum_and_avg_require_a_numeric_type() {
        assert!(AggregateFn::Sum.requires_numeric());
        assert!(AggregateFn::Avg.requires_numeric());
        // Ordering works on any comparable column, so these stay open.
        assert!(!AggregateFn::Min.requires_numeric());
        assert!(!AggregateFn::Max.requires_numeric());
    }

    #[test]
    fn recognises_the_numeric_postgres_types() {
        for t in ["numeric", "NUMERIC(14,2)", "integer", "bigint", "double precision", "money"] {
            assert!(is_numeric_pg_type(t), "{t} should count as numeric");
        }
        for t in ["text", "uuid", "timestamptz", "boolean", "jsonb", "USER-DEFINED"] {
            assert!(!is_numeric_pg_type(t), "{t} must not accept a SUM");
        }
    }
}

#[cfg(test)]
mod filter_cast_tests {
    use super::*;

    #[test]
    fn typed_base_columns_get_their_own_cast() {
        for (t, want) in [
            ("boolean", "boolean"),
            ("integer", "integer"),
            ("bigint", "bigint"),
            ("smallint", "smallint"),
            ("numeric", "numeric"),
            ("double precision", "double precision"),
            ("uuid", "uuid"),
            ("date", "date"),
            ("time without time zone", "time"),
            ("timestamp with time zone", "timestamptz"),
            ("timestamp without time zone", "timestamp"),
        ] {
            assert_eq!(filter_cast_for(t, "b").as_deref(), Some(want), "{t}");
        }
    }

    #[test]
    fn text_like_and_composite_columns_keep_the_text_bind() {
        for t in ["text", "character varying", "character", "jsonb", "json", "bytea", "text[]", "uuid[]"] {
            assert_eq!(filter_cast_for(t, "b"), None, "{t}");
        }
        // A domain or composite is not an enum, even though both are user-defined.
        assert_eq!(filter_cast_for("approvals.money_amount", "d"), None);
        assert_eq!(filter_cast_for("approvals.address", "c"), None);
    }

    #[test]
    fn an_enum_casts_to_its_catalog_name() {
        assert_eq!(filter_cast_for("approval_status", "e").as_deref(), Some("approval_status"));
        assert_eq!(
            filter_cast_for("recruitment.stage_kind", "e").as_deref(),
            Some("recruitment.stage_kind")
        );
    }

    #[test]
    fn a_generated_hint_wins_over_the_catalog_for_its_column() {
        let hints: HashMap<String, String> =
            [("id", "uuid"), ("status", "approval_status")].iter().map(|(k, v)| (k.to_string(), v.to_string())).collect();
        let catalog: HashMap<String, String> = [
            ("id", "uuid"),
            ("status", "approvals.approval_status"),
            ("folded", "boolean"),
            ("requested_by", "uuid"),
        ]
        .iter()
        .map(|(k, v)| (k.to_string(), v.to_string()))
        .collect();
        let merged = merge_filter_casts(&hints, catalog);
        // The bare enum hint gives way to the catalog's qualified name for the same type: a
        // tenant's search path need not hold the module's schema.
        assert_eq!(merged["status"], "approvals.approval_status");
        assert_eq!(merged["id"], "uuid");
        assert_eq!(merged["folded"], "boolean");
        assert_eq!(merged["requested_by"], "uuid");
        assert_eq!(merged.len(), 4);
    }

    #[test]
    fn a_hint_for_another_type_still_decides_its_column() {
        let hints: HashMap<String, String> =
            [("amount", "numeric")].iter().map(|(k, v)| (k.to_string(), v.to_string())).collect();
        let catalog: HashMap<String, String> =
            [("amount", "real")].iter().map(|(k, v)| (k.to_string(), v.to_string())).collect();
        assert_eq!(merge_filter_casts(&hints, catalog)["amount"], "numeric");
    }

    #[test]
    fn a_filter_on_a_bare_enum_hint_reads_the_catalog_to_qualify_it() {
        let hints: HashMap<String, String> = [("id", "uuid"), ("status", "task_status")]
            .iter()
            .map(|(k, v)| (k.to_string(), v.to_string()))
            .collect();
        let needs = |pairs: &[(&str, &str)]| {
            let f: HashMap<String, String> =
                pairs.iter().map(|(k, v)| (k.to_string(), v.to_string())).collect();
            parse_query_filter(&f, &hints, None).unwrap().needs_catalog_types()
        };
        assert!(needs(&[("status[in]", "done,skipped")]));
        assert!(needs(&[("status[eq]", "done")]));
        // A uuid hint resolves anywhere; nothing to read.
        assert!(!needs(&[("id[in]", "a,b")]));
    }

    #[test]
    fn a_sort_on_an_enum_casts_to_its_schema_qualified_name() {
        assert_eq!(
            cast_suffix("USER-DEFINED", "task_status", "lifecycle").as_deref(),
            Some("lifecycle.task_status")
        );
        assert_eq!(cast_suffix("USER-DEFINED", "mood", "public").as_deref(), Some("mood"));
        assert_eq!(cast_suffix("uuid", "uuid", "pg_catalog").as_deref(), Some("uuid"));
    }

    #[test]
    fn only_a_filter_with_an_uncast_comparison_reads_the_catalog() {
        let hints: HashMap<String, String> =
            [("id", "uuid")].iter().map(|(k, v)| (k.to_string(), v.to_string())).collect();
        let parse = |pairs: &[(&str, &str)]| {
            let f: HashMap<String, String> =
                pairs.iter().map(|(k, v)| (k.to_string(), v.to_string())).collect();
            parse_query_filter(&f, &hints, None).unwrap().has_uncast_value_conditions()
        };
        assert!(!parse(&[("id[in]", "a,b"), ("name[contain]", "x"), ("limit", "5")]));
        assert!(!parse(&[("deleted_by[isnull]", "1")]));
        assert!(parse(&[("folded[eq]", "false")]));
        assert!(parse(&[("folded", "false")]));
        assert!(parse(&[("scheduled_at[between]", "2026-10-01,2026-10-03")]));
        assert!(parse(&[("sequence[or]", "3")]));
    }
}

#[cfg(test)]
mod keyset_predicate_tests {
    use super::and_keyset_predicate;

    #[test]
    fn a_client_or_cannot_escape_the_keyset_window() {
        assert_eq!(
            and_keyset_predicate(" WHERE kind = $1 OR status = $2", "id > $3"),
            " WHERE (kind = $1 OR status = $2) AND (id > $3)"
        );
    }

    #[test]
    fn no_filter_leaves_the_keyset_alone() {
        assert_eq!(and_keyset_predicate("", "id > $1"), " WHERE id > $1");
    }
}

#[cfg(test)]
mod changed_columns_tests {
    use super::changed_columns;

    #[derive(serde::Serialize)]
    struct Row {
        id: &'static str,
        notes: &'static str,
        status: &'static str,
        #[serde(skip_serializing_if = "Option::is_none")]
        label: Option<&'static str>,
    }

    #[test]
    fn only_the_fields_that_differ_are_changed() {
        let before = Row {
            id: "1",
            notes: "a",
            status: "draft",
            label: None,
        };
        let after = Row {
            id: "2",
            notes: "b",
            status: "draft",
            label: Some("x"),
        };
        assert_eq!(
            changed_columns(&before, &after).unwrap(),
            vec!["label".to_string(), "notes".to_string()]
        );
        assert!(changed_columns(&before, &before).unwrap().is_empty());
    }
}

#[cfg(test)]
mod total_order_tests {
    use super::{order_by_sql, total_order};
    use crate::filter::SortDirection::{Asc, Desc};

    fn s(f: &str, d: crate::filter::SortDirection) -> (String, crate::filter::SortDirection) {
        (f.to_string(), d)
    }

    #[test]
    fn no_sort_orders_by_the_key_ascending() {
        assert_eq!(order_by_sql(&total_order(vec![]), false), " ORDER BY id ASC");
    }

    #[test]
    fn a_sort_gets_the_key_as_its_final_tiebreaker() {
        let order = total_order(vec![s("status", Desc), s("name", Asc)]);
        assert_eq!(order_by_sql(&order, false), " ORDER BY status DESC, name ASC, id ASC");
    }

    #[test]
    fn the_tiebreaker_is_not_repeated_when_the_key_is_already_sorted() {
        let last = total_order(vec![s("name", Asc), s("id", Desc)]);
        assert_eq!(order_by_sql(&last, false), " ORDER BY name ASC, id DESC");
        let first = total_order(vec![s("id", Desc), s("name", Asc)]);
        assert_eq!(order_by_sql(&first, false), " ORDER BY id DESC, name ASC");
    }

    #[test]
    fn a_backwards_walk_inverts_every_direction_including_the_key() {
        let order = total_order(vec![s("created_at", Desc)]);
        assert_eq!(order_by_sql(&order, true), " ORDER BY created_at ASC, id DESC");
    }

    #[test]
    fn the_order_is_idempotent() {
        let once = total_order(vec![s("name", Asc)]);
        assert_eq!(total_order(once.clone()), once);
    }
}