cratestack_sqlx/query/read/
find_many.rs

1//! `find_many` — typed multi-row read with filter / order /
2//! pagination / `FOR UPDATE`. The `preview_*_sql` previews live in
3//! [`super::find_many_preview`] to keep this file under budget.
4
5use cratestack_core::{CratestackContext, CratestackError};
6use cratestack_sql::ReadSource;
7
8use crate::query::support::{ReadPolicyKind, push_order_and_paging, push_scoped_conditions};
9use crate::{FilterExpr, OrderClause, SqlxRuntime, sqlx};
10
11#[derive(Clone)]
12pub struct FindMany<'a, M: 'static, PK: 'static> {
13    pub(crate) runtime: &'a SqlxRuntime,
14    /// Either a `&'static ModelDescriptor<M, PK>` (typical) or a
15    /// `&'static ViewDescriptor<M, PK>` (view path). Both impl
16    /// `ReadSource<M, PK>`.
17    pub(crate) descriptor: &'static dyn ReadSource<M, PK>,
18    pub(crate) filters: Vec<FilterExpr>,
19    pub(crate) order_by: Vec<OrderClause>,
20    pub(crate) limit: Option<i64>,
21    pub(crate) offset: Option<i64>,
22    pub(crate) for_update: bool,
23}
24
25impl<'a, M: 'static, PK: 'static> FindMany<'a, M, PK> {
26    pub fn where_(mut self, filter: crate::Filter) -> Self {
27        self.filters.push(FilterExpr::from(filter));
28        self
29    }
30
31    pub fn where_expr(mut self, filter: FilterExpr) -> Self {
32        self.filters.push(filter);
33        self
34    }
35
36    pub fn where_any(mut self, filters: impl IntoIterator<Item = FilterExpr>) -> Self {
37        self.filters.push(FilterExpr::any(filters));
38        self
39    }
40
41    /// Conditionally append a filter. `None` is a no-op so callers can
42    /// pipe `FieldRef::match_optional(...)` results straight in
43    /// without an `if let` ladder at every optional-param site.
44    pub fn where_optional<F>(mut self, filter: Option<F>) -> Self
45    where
46        F: Into<FilterExpr>,
47    {
48        if let Some(filter) = filter {
49            self.filters.push(filter.into());
50        }
51        self
52    }
53
54    pub fn order_by(mut self, clause: OrderClause) -> Self {
55        self.order_by.push(clause);
56        self
57    }
58
59    pub fn limit(mut self, limit: i64) -> Self {
60        self.limit = Some(limit);
61        self
62    }
63
64    pub fn offset(mut self, offset: i64) -> Self {
65        self.offset = Some(offset);
66        self
67    }
68
69    /// Emit `SELECT ... FOR UPDATE` so the engine takes an exclusive
70    /// row-level lock on every matched row for the surrounding
71    /// transaction. Only meaningful when paired with [`Self::run_in_tx`].
72    pub fn for_update(mut self) -> Self {
73        self.for_update = true;
74        self
75    }
76
77    /// The query's filters, ordering and paging with **no** authorization
78    /// scope: neither this model's read policy nor, inside relation
79    /// filter/sort subqueries, the related model's. Use
80    /// [`Self::preview_scoped_sql`] to see what actually executes.
81    pub fn preview_sql(&self) -> String {
82        super::find_many_preview::preview_sql(self)
83    }
84
85    /// The query as it executes for `ctx`, including the related read
86    /// scope inside every relation filter/sort subquery.
87    pub fn preview_scoped_sql(&self, ctx: &CratestackContext) -> String {
88        super::find_many_preview::preview_scoped_sql(self, ctx)
89    }
90
91    pub async fn run(self, ctx: &CratestackContext) -> Result<Vec<M>, CratestackError>
92    where
93        for<'r> M: Send + Unpin + sqlx::FromRow<'r, sqlx::postgres::PgRow>,
94    {
95        // Inside an `@isolation` procedure: on its transaction (procedure-isolation.md §4).
96        if let Some(bound) = self.runtime.bound() {
97            return crate::bound::in_bound_savepoint!(bound, |sp| self.run_in_tx(sp, ctx));
98        }
99        let order_by = self.effective_order_by();
100        let mut query = sqlx::QueryBuilder::<sqlx::Postgres>::new("SELECT ");
101        query
102            .push(self.descriptor.select_projection())
103            .push(" FROM ")
104            .push(self.descriptor.table_name());
105
106        push_scoped_conditions(
107            &mut query,
108            self.descriptor,
109            &self.filters,
110            None::<(&'static str, i64)>,
111            ctx,
112            ReadPolicyKind::List,
113        );
114        push_order_and_paging(&mut query, &order_by, self.limit, self.offset, ctx);
115        if self.for_update {
116            query.push(" FOR UPDATE");
117        }
118
119        query
120            .build_query_as::<M>()
121            .fetch_all(self.runtime.pool())
122            .await
123            .map_err(crate::error::cratestack_error_from_sqlx)
124    }
125
126    /// Run inside a caller-supplied transaction. Required when pairing
127    /// with [`Self::for_update`].
128    pub async fn run_in_tx<'tx>(
129        self,
130        tx: &mut sqlx::Transaction<'tx, sqlx::Postgres>,
131        ctx: &CratestackContext,
132    ) -> Result<Vec<M>, CratestackError>
133    where
134        for<'r> M: Send + Unpin + sqlx::FromRow<'r, sqlx::postgres::PgRow>,
135    {
136        let order_by = self.effective_order_by();
137        let mut query = sqlx::QueryBuilder::<sqlx::Postgres>::new("SELECT ");
138        query
139            .push(self.descriptor.select_projection())
140            .push(" FROM ")
141            .push(self.descriptor.table_name());
142
143        push_scoped_conditions(
144            &mut query,
145            self.descriptor,
146            &self.filters,
147            None::<(&'static str, i64)>,
148            ctx,
149            ReadPolicyKind::List,
150        );
151        push_order_and_paging(&mut query, &order_by, self.limit, self.offset, ctx);
152        if self.for_update {
153            query.push(" FOR UPDATE");
154        }
155
156        query
157            .build_query_as::<M>()
158            .fetch_all(&mut **tx)
159            .await
160            .map_err(crate::error::cratestack_error_from_sqlx)
161    }
162
163    pub(super) fn effective_order_by(&self) -> Vec<OrderClause> {
164        let mut order_by = self.order_by.clone();
165        let Some(direction) = order_by
166            .iter()
167            .find(|clause| clause.is_relation_scalar())
168            .map(OrderClause::direction)
169        else {
170            return order_by;
171        };
172
173        if order_by
174            .iter()
175            .any(|clause| clause.targets_column(self.descriptor.primary_key()))
176        {
177            return order_by;
178        }
179
180        order_by.push(OrderClause::column(
181            self.descriptor.primary_key(),
182            direction,
183        ));
184        order_by
185    }
186
187    /// Side-load a to-one relation alongside the matched rows. Two
188    /// queries, not a SQL JOIN, so the related-side read policy +
189    /// soft-delete inherit from `find_many` for free.
190    pub fn include<Rel, RelPK>(
191        self,
192        relation: cratestack_sql::RelationInclude<M, Rel, RelPK>,
193    ) -> super::find_many_with::FindManyWith<'a, M, PK, Rel, RelPK>
194    where
195        Rel: 'static,
196        RelPK: 'static,
197    {
198        super::find_many_with::FindManyWith::new(self, relation)
199    }
200}
cratestack_sqlx/query/read/find_many.rs

cratestack_sqlx/query/read/
find_many.rs
