umbral_core/orm/queryset/mod.rs
1//! `QuerySet<T>` and `Manager<T>`: chainable lazy SQL builder + entry
2//! point.
3//!
4//! `T::objects()` returns a `Manager<T>`; chaining `.filter` / `.order_by`
5//! / `.limit` / etc. on it (or on a `QuerySet<T>` directly) yields a new
6//! `QuerySet<T>`. Terminals (`.fetch`, `.first`, `.count`, `.exists`)
7//! await an async DB roundtrip via the ambient or explicit pool.
8//!
9//! At M1 the surface is intentionally narrow per
10//! `docs/specs/03-orm-querysets.md`: filter / order_by / limit / offset
11//! for chaining, and fetch / first / count / exists for terminals. No
12//! exclude / distinct / values / annotate / aggregate / update / delete
13//! yet — those land as later milestones surface real need.
14//!
15//! M2 lifted the terminals and the `Manager` delegation onto a generic
16//! `T: Model` bound. The table name comes from `T::TABLE`, the SELECT
17//! column list from `T::FIELDS`, and row materialisation from the
18//! `FromRow` bound the terminals carry. M3 generates the `Model` impl
19//! from `#[derive(Model)]`.
20//!
21//! ## Phase 2.5 — backend-agnostic terminals
22//!
23//! Through Phase 2 the QuerySet stored a `SqlitePool` and built every
24//! query with sea-query's `SqliteQueryBuilder`. Phase 2.5 widens that:
25//! the explicit-pool slot is `Option<DbPool>`, `.on(&SqlitePool)` keeps
26//! working unchanged, and a new `.on_pg(&PgPool)` registers a Postgres
27//! pool. The terminal methods dispatch on the resolved pool variant —
28//! SQLite path uses `SqliteQueryBuilder` + a `SqlitePool` executor;
29//! Postgres path uses `PostgresQueryBuilder` + a `PgPool` executor.
30//!
31//! The row-materialization bound on each terminal is the conjunction
32//! of both backends' `FromRow` impls. `#[derive(sqlx::FromRow)]` emits
33//! a generic-over-`R` impl, so a user struct with standard field
34//! types satisfies both bounds without any per-backend ceremony.
35
36mod backend_pg;
37mod backend_sqlite;
38mod errors;
39pub(crate) mod hydration;
40mod tx;
41mod write_helpers;
42
43pub use errors::{GetError, TryForEachError};
44use hydration::{hydrate_prefetch_related, hydrate_select_related};
45pub use tx::QuerySetTx;
46use write_helpers::{
47 build_insert_many_for, build_insert_one_for, fk_pk_hint, pk_field, serialize_to_map,
48};
49
50use std::collections::HashMap;
51use std::marker::PhantomData;
52
53use sea_query::{
54 Alias, Expr, Func, IntoIden, Order, PostgresQueryBuilder, Query, SqliteQueryBuilder,
55};
56use sea_query_binder::SqlxBinder;
57use serde_json::Value as JsonValue;
58
59use crate::db::DbPool;
60use crate::orm::{FExpr, HydrateRelated, Model, OrderExpr, Predicate};
61use umbral_casing::to_snake_case;
62
63/// Entry point for queries on a model.
64///
65/// `Manager<T>` wraps a freshly-constructed `QuerySet<T>` and exposes
66/// the same chainable surface. The user never constructs one directly;
67/// `Post::objects()` is the only door.
68pub struct Manager<T> {
69 _phantom: PhantomData<T>,
70 /// Per-Manager override for the `atomic_transactions` builder
71 /// default. `None` = inherit the global default; `Some(true)` =
72 /// wrap subsequent writes in a transaction; `Some(false)` =
73 /// explicitly opt out. Propagates into the QuerySet `queryset()`
74 /// constructs.
75 atomic: Option<bool>,
76}
77
78impl<T> Manager<T> {
79 pub(crate) fn new() -> Self {
80 Self {
81 _phantom: PhantomData,
82 atomic: None,
83 }
84 }
85
86 /// Wrap every write terminal that hangs off this Manager in a
87 /// transaction. Equivalent to calling `.atomic()` on each
88 /// QuerySet derived from it. Per-call `.non_atomic()` overrides.
89 pub fn atomic(mut self) -> Self {
90 self.atomic = Some(true);
91 self
92 }
93
94 /// Opt this Manager (and every QuerySet derived from it) out of
95 /// the global `App::builder().atomic_transactions(true)` default.
96 pub fn non_atomic(mut self) -> Self {
97 self.atomic = Some(false);
98 self
99 }
100}
101
102impl<T> Default for Manager<T> {
103 fn default() -> Self {
104 Self::new()
105 }
106}
107
108/// SQL join flavor recorded per `join_related` hop. `None` in a
109/// `JoinReq` means "infer from FK nullability" (gap 4c); an explicit
110/// `left_/inner_/right_join_related` records `Some(..)`.
111#[derive(Debug, Clone, Copy, PartialEq, Eq)]
112pub enum JoinKind {
113 Inner,
114 Left,
115 Right,
116}
117
118impl JoinKind {
119 /// Lower to sea-query's join type.
120 pub(crate) fn sea(self) -> sea_query::JoinType {
121 match self {
122 JoinKind::Inner => sea_query::JoinType::InnerJoin,
123 JoinKind::Left => sea_query::JoinType::LeftJoin,
124 JoinKind::Right => sea_query::JoinType::RightJoin,
125 }
126 }
127}
128
129/// One requested eager-join: a dotted relation path (`"plugin__author"`)
130/// plus the join type to apply to the LAST hop. `kind: None` means
131/// auto-infer per-hop from FK nullability (INNER for NOT NULL, LEFT for
132/// nullable), the default. The explicit methods pin `Some(..)`.
133#[derive(Debug, Clone)]
134pub(crate) struct JoinReq {
135 pub(crate) path: String,
136 pub(crate) kind: Option<JoinKind>,
137}
138
139/// A lazy, chainable SQL query.
140///
141/// Carries a sea-query `SelectStatement` plus pool-resolution state.
142/// Nothing is sent to the database until a terminal method is awaited.
143/// Cloning is cheap (the `SelectStatement` clones in O(query size)).
144pub struct QuerySet<T> {
145 /// The base SelectStatement — FROM, columns, joins, group-by,
146 /// order-by, limit, offset. Filters are NOT applied here; they
147 /// accumulate on [`Self::predicates`] and get woven in at
148 /// terminal time, so per-backend predicate variants (Phase
149 /// 4.2.2) can pick the right SimpleExpr based on the resolved
150 /// pool.
151 pub(crate) query: sea_query::SelectStatement,
152 /// Accumulated filter predicates. Each one renders to either its
153 /// default `cond` (for Postgres) or its `cond_sqlite` override
154 /// (for SQLite, if set) at terminal time.
155 pub(crate) predicates: Vec<Predicate<T>>,
156 pub(crate) explicit_pool: Option<DbPool>,
157 /// FK field names requested for eager loading via `select_related`.
158 /// After the main query returns rows, a batch `IN (...)` query
159 /// fetches the related rows for each named field and calls
160 /// `HydrateRelated::hydrate_fk` to populate `ForeignKey.resolved`.
161 pub(crate) select_related: Vec<String>,
162 /// M2M field names requested for eager loading via
163 /// `prefetch_related`. After the main query, one batched JOIN
164 /// against the junction + child table fetches every related row
165 /// for every parent in a single round-trip; each parent's
166 /// `M2M.resolved` slot is populated via
167 /// `HydrateRelated::set_m2m_resolved_json`. Gap #19.
168 pub(crate) prefetch_related: Vec<String>,
169 /// BUG-8: `#[umbral(ordering = [...])]` lowers to a default ORDER
170 /// BY applied at terminal time when the caller didn't supply an
171 /// explicit `.order_by(...)`. The semantics: explicit calls REPLACE
172 /// the default rather than appending to it.
173 pub(crate) default_ordering: Vec<(&'static str, bool)>,
174 /// Set to `true` the first time `.order_by(...)` is called; when
175 /// `false`, `build_query_for` applies `default_ordering`.
176 pub(crate) explicit_order: bool,
177 /// Per-QuerySet override for the `atomic_transactions` builder
178 /// default. `None` = inherit the global default via
179 /// [`crate::db::atomic_default`]; `Some(true)` = wrap this
180 /// QuerySet's write terminal in a transaction; `Some(false)` =
181 /// explicitly opt out.
182 pub(crate) atomic: Option<bool>,
183 /// Feature #72 — soft-delete state. Snapshotted from
184 /// `T::SOFT_DELETE` when the QuerySet is constructed from a
185 /// Manager (the no-bounds `QuerySet::new` constructor leaves
186 /// this `false` so hand-built QuerySets stay opt-out by
187 /// default).
188 pub(crate) soft_delete_active: bool,
189 /// True when the caller opted back into soft-deleted rows via
190 /// `.with_deleted()`. Skips the auto `WHERE deleted_at IS NULL`
191 /// injection.
192 pub(crate) with_deleted: bool,
193 /// True when the caller wants ONLY soft-deleted rows via
194 /// `.only_deleted()`. Inverts the auto-filter to
195 /// `WHERE deleted_at IS NOT NULL`.
196 pub(crate) only_deleted: bool,
197 /// True when the caller asked for a real DELETE via
198 /// `.hard_delete()` — bypasses the soft-delete rewrite that
199 /// would normally turn `delete()` into an UPDATE.
200 pub(crate) hard_delete: bool,
201 /// Gap #111 — column projection set by [`Self::only`]. When
202 /// `Some`, [`Self::to_sql`] / [`Self::to_sql_pg`] swap the
203 /// SELECT list for just these columns, and the typed terminals
204 /// (`fetch` / `first` / `get`) refuse to run with a clear error
205 /// pointing the caller at [`Self::values`] (FromRow can't
206 /// hydrate `T` from a partial-column row). `None` keeps the
207 /// pre-#111 behaviour (full SELECT).
208 pub(crate) only_cols: Option<Vec<String>>,
209 /// FK field names requested for JOIN-based prefetch via
210 /// [`Self::join_related`]. Distinct from `select_related`:
211 /// `join_related` weaves `LEFT JOIN <related_table>` into the
212 /// main SELECT (with aliased columns `<field>__<col>`) so one
213 /// round-trip pulls parent + related rows together. The existing
214 /// `select_related` path keeps its "batched-IN-followup-query"
215 /// shape — both are valid; this one wins when round-trip count
216 /// matters more than the per-row column overhead.
217 pub(crate) join_related: Vec<JoinReq>,
218 /// Related-aggregate annotations added via
219 /// [`Self::annotate_related`] / [`Self::annotate_count`]. Applied
220 /// inside `build_query_for`, so EVERY terminal and introspection
221 /// path — `fetch_annotated`, `explain`, `to_sql`, `to_sql_pg` —
222 /// sees the same correlated subqueries. That's the
223 /// `annotate()` contract: an annotation is query-builder state,
224 /// not a side query.
225 pub(crate) annotations: Vec<RelatedAnnotation>,
226 /// gaps3 #29 — `ORDER BY <annotation alias>`. Kept separate from
227 /// `order_by`, which is typed against the model's own columns and so cannot
228 /// name a computed alias.
229 pub(crate) annotation_order: Vec<(String, bool)>,
230 /// audit_2 plugin-storage-tasks #6 — when `true`, a read terminal appends
231 /// `FOR UPDATE SKIP LOCKED` (Postgres only). Lets N contending workers each
232 /// claim a DIFFERENT row instead of all piling onto the same head row and
233 /// serializing on its lock. A no-op on SQLite (no such clause; its
234 /// single-writer model needs none). Set via [`Self::for_update_skip_locked`].
235 pub(crate) for_update_skip_locked: bool,
236 /// gaps4 #27: the caller-supplied `.limit(n)`, tracked separately from the
237 /// sea-query statement (which doesn't expose reading its own LIMIT back).
238 /// `try_for_each` needs it: without it, chunked iteration overwrote the
239 /// statement's LIMIT with its own paging value and silently walked the
240 /// whole table for a caller who had bounded the scan.
241 pub(crate) user_limit: Option<u64>,
242 /// A caller-supplied `.offset(n)`, tracked separately from `self.query`
243 /// because sea-query won't hand its OFFSET back — `try_for_each` needs it
244 /// to start paging from the requested row instead of 0.
245 pub(crate) user_offset: Option<u64>,
246 _phantom: PhantomData<T>,
247}
248
249// Manual `Clone` — NOT `#[derive(Clone)]`, because the derive would force a
250// spurious `T: Clone` bound (every field is `T`-independent or carries its
251// own `T`-free `Clone`: `Predicate<T>` has a manual `impl<T> Clone`, and
252// `PhantomData<T>` clones for any `T`). The doc comment above already
253// promised cheap cloning; this is what makes `Paginator` (and any
254// requery-without-consume caller) slice the same query per page.
255impl<T> Clone for QuerySet<T> {
256 fn clone(&self) -> Self {
257 Self {
258 query: self.query.clone(),
259 predicates: self.predicates.clone(),
260 explicit_pool: self.explicit_pool.clone(),
261 select_related: self.select_related.clone(),
262 prefetch_related: self.prefetch_related.clone(),
263 default_ordering: self.default_ordering.clone(),
264 explicit_order: self.explicit_order,
265 atomic: self.atomic,
266 soft_delete_active: self.soft_delete_active,
267 with_deleted: self.with_deleted,
268 only_deleted: self.only_deleted,
269 hard_delete: self.hard_delete,
270 only_cols: self.only_cols.clone(),
271 join_related: self.join_related.clone(),
272 annotations: self.annotations.clone(),
273 annotation_order: self.annotation_order.clone(),
274 user_limit: self.user_limit,
275 user_offset: self.user_offset,
276 for_update_skip_locked: self.for_update_skip_locked,
277 _phantom: PhantomData,
278 }
279 }
280}
281
282// Manual `Debug` for the same `T: Debug`-free reason; `Predicate<T>`'s own
283// `Debug` is likewise `T`-free.
284impl<T> std::fmt::Debug for QuerySet<T> {
285 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
286 // `Predicate<T>` is intentionally not `Debug` (its `SimpleExpr`
287 // carries no `T`), so report the predicate count rather than the
288 // opaque expressions.
289 f.debug_struct("QuerySet")
290 .field("query", &self.query)
291 .field(
292 "predicates",
293 &format_args!("[{} predicate(s)]", self.predicates.len()),
294 )
295 .field("explicit_pool", &self.explicit_pool)
296 .field("select_related", &self.select_related)
297 .field("prefetch_related", &self.prefetch_related)
298 .field("default_ordering", &self.default_ordering)
299 .field("explicit_order", &self.explicit_order)
300 .field("atomic", &self.atomic)
301 .field("soft_delete_active", &self.soft_delete_active)
302 .field("with_deleted", &self.with_deleted)
303 .field("only_deleted", &self.only_deleted)
304 .field("hard_delete", &self.hard_delete)
305 .field("only_cols", &self.only_cols)
306 .field("join_related", &self.join_related)
307 .field("annotations", &self.annotations)
308 .finish()
309 }
310}
311
312/// One related-aggregate annotation on a [`QuerySet`] —
313/// `(alias, relation, aggregate)` resolved against the model's
314/// `REVERSE_FK_RELATIONS` at builder time. A name that fails to
315/// resolve is stored poisoned (`resolved: Err`) so the infallible
316/// builder stays chainable while every fallible consumer
317/// (`fetch_annotated`, `explain`) reports it loudly.
318#[derive(Debug, Clone)]
319pub(crate) struct RelatedAnnotation {
320 pub(crate) alias: String,
321 pub(crate) agg: crate::orm::Aggregate,
322 /// `Ok((child_table, fk_column, parent_table, parent_pk))` or the
323 /// loud error message for an unknown relation.
324 pub(crate) resolved: Result<(String, String, String, String), String>,
325 /// Child model is `#[umbral(soft_delete)]` — fold
326 /// `AND <child>.deleted_at IS NULL` into the correlated subquery so
327 /// a trashed child stops inflating the parent's count.
328 pub(crate) child_soft_delete: bool,
329 /// Optional child-side predicate (a filtered count),
330 /// pre-rendered to a backend-default `SimpleExpr`. ANDed into the
331 /// subquery WHERE. From `annotate_count_where`.
332 pub(crate) child_filter: Option<sea_query::SimpleExpr>,
333 /// `Some(junction_table)` when this annotation counts M2M junction
334 /// rows instead of child rows (`annotate_count` over an `M2M<T>`).
335 pub(crate) m2m_junction: Option<String>,
336}
337
338/// Outcome of auto-discovering a reverse-FK relation (gaps2 #45) when
339/// the parent declares no matching `ReverseSet` field. The resolver
340/// scans the registry for children whose FK targets the parent table
341/// and matches `relation` against their conventional name forms.
342enum AutoDiscovery {
343 /// Exactly one (child, fk_column) candidate matched.
344 Resolved {
345 child_table: String,
346 fk_column: String,
347 soft_delete: bool,
348 },
349 /// Two or more candidates matched — the caller must declare a
350 /// `#[umbral(reverse_fk = "...")]` field to disambiguate. Carries
351 /// the candidate `child.fk` labels for the error message.
352 Ambiguous(Vec<String>),
353 /// No candidate matched. Carries the list of auto-discoverable
354 /// child names so the error can teach the available relations.
355 NotFound(Vec<String>),
356}
357
358// `snake_case` replaced by `umbral_casing::to_snake_case` (imported above)
359// in the gaps2 #77 consolidation refactor.
360
361/// Scan the model registry for children whose FK targets `T::TABLE`,
362/// and match `relation` against each candidate's conventional name
363/// forms: the child's table name, the child's struct name in
364/// snake_case and bare-lowercase, and any of those with a `_set`
365/// suffix (the `<model>_set` form). Declared `REVERSE_FK_RELATIONS` /
366/// `M2M_RELATIONS` are resolved by the caller BEFORE this runs, so
367/// they always take precedence.
368fn discover_reverse_relation<T: crate::orm::Model>(relation: &str) -> AutoDiscovery {
369 if !crate::migrate::is_initialised() {
370 return AutoDiscovery::NotFound(Vec::new());
371 }
372 let parent_table = T::TABLE;
373 // Each candidate: (child_table, fk_column, child_soft_delete).
374 let mut candidates: Vec<(String, String, bool)> = Vec::new();
375 let mut discoverable: Vec<String> = Vec::new();
376 for meta in crate::migrate::registered_models() {
377 for col in &meta.fields {
378 if col.fk_target.as_deref() != Some(parent_table) {
379 continue;
380 }
381 // Conventional name forms this (child, fk_column) answers to.
382 let snake = to_snake_case(&meta.name);
383 let lower = meta.name.to_ascii_lowercase();
384 let mut forms = vec![
385 meta.table.clone(),
386 snake.clone(),
387 lower.clone(),
388 format!("{}_set", meta.table),
389 format!("{snake}_set"),
390 format!("{lower}_set"),
391 ];
392 forms.sort();
393 forms.dedup();
394 // Surface a friendly name for "available children" errors.
395 discoverable.push(format!("{}_set", meta.table));
396 if forms.iter().any(|f| f == relation) {
397 candidates.push((meta.table.clone(), col.name.clone(), meta.soft_delete));
398 }
399 }
400 }
401 match candidates.len() {
402 1 => {
403 let (child_table, fk_column, soft_delete) = candidates.pop().unwrap();
404 AutoDiscovery::Resolved {
405 child_table,
406 fk_column,
407 soft_delete,
408 }
409 }
410 0 => {
411 discoverable.sort();
412 discoverable.dedup();
413 AutoDiscovery::NotFound(discoverable)
414 }
415 _ => {
416 let labels = candidates
417 .into_iter()
418 .map(|(child, fk, _)| format!("{child}.{fk}"))
419 .collect();
420 AutoDiscovery::Ambiguous(labels)
421 }
422 }
423}
424
425impl<T> QuerySet<T> {
426 pub(crate) fn new(query: sea_query::SelectStatement) -> Self {
427 Self {
428 query,
429 default_ordering: Vec::new(),
430 explicit_order: false,
431 predicates: Vec::new(),
432 explicit_pool: None,
433 select_related: Vec::new(),
434 prefetch_related: Vec::new(),
435 atomic: None,
436 soft_delete_active: false,
437 with_deleted: false,
438 only_deleted: false,
439 hard_delete: false,
440 only_cols: None,
441 join_related: Vec::new(),
442 annotations: Vec::new(),
443 annotation_order: Vec::new(),
444 for_update_skip_locked: false,
445 user_limit: None,
446 user_offset: None,
447 _phantom: PhantomData,
448 }
449 }
450
451 /// Feature #72 — include soft-deleted rows in this query. Skips
452 /// the auto `WHERE deleted_at IS NULL` injection. No-op on
453 /// models that aren't tagged `#[umbral(soft_delete)]`.
454 pub fn with_deleted(mut self) -> Self {
455 self.with_deleted = true;
456 self
457 }
458
459 /// Feature #72 — only soft-deleted rows. Useful for admin
460 /// trash views and undelete workflows. No-op on models that
461 /// aren't tagged `#[umbral(soft_delete)]`.
462 pub fn only_deleted(mut self) -> Self {
463 self.only_deleted = true;
464 self
465 }
466
467 /// Feature #72 — force a real DELETE for the next `.delete()`
468 /// terminal call. Soft-delete models normally rewrite delete()
469 /// as `UPDATE ... SET deleted_at = NOW()`; `.hard_delete()`
470 /// bypasses that for GDPR purges, test cleanup, or any other
471 /// case where the row truly should be gone. No-op on models
472 /// that aren't tagged `#[umbral(soft_delete)]` (their delete()
473 /// is already a hard DELETE).
474 pub fn hard_delete(mut self) -> Self {
475 self.hard_delete = true;
476 self
477 }
478
479 /// Gap #111 — restrict the SELECT to the named columns.
480 ///
481 /// Affects [`Self::to_sql`] / [`Self::to_sql_pg`] (the SELECT list
482 /// shrinks to just these columns) and propagates into
483 /// [`Self::values`] when that terminal is called without its own
484 /// explicit column slice.
485 ///
486 /// **The typed terminals (`fetch` / `first` / `get`) refuse to
487 /// run with `.only()` set** because a partial-column row can't
488 /// satisfy `T`'s `FromRow` impl. The error message points at
489 /// `.values(...)` (returns `Vec<serde_json::Value>`) as the
490 /// execution path. Use `.only()` for `.to_sql()` inspection and
491 /// `.values()` for actual reads.
492 ///
493 /// ```rust,ignore
494 /// // Inspect: "SELECT \"id\", \"name\" FROM \"brand\" WHERE \"id\" = ?"
495 /// let sql = Brand::objects()
496 /// .filter(brand::ID.eq(1))
497 /// .only(&["id", "name"])
498 /// .to_sql();
499 ///
500 /// // Execute (returns JSON rows):
501 /// let rows = Brand::objects()
502 /// .filter(brand::ID.eq(1))
503 /// .values(&["id", "name"])
504 /// .await?;
505 /// ```
506 ///
507 /// Unknown column names are not validated here — they surface at
508 /// terminal time the same way `.values()` reports them (the
509 /// rendered SQL contains the bad identifier and SQLite/Postgres
510 /// raises). This keeps the chainable surface return-type-stable.
511 pub fn only(mut self, cols: &[&str]) -> Self {
512 self.only_cols = Some(cols.iter().map(|s| s.to_string()).collect());
513 self
514 }
515
516 /// Wrap this QuerySet's write terminal in a transaction. Reads are
517 /// unaffected (read terminals are single statements and the DB
518 /// gives them a consistent snapshot). Mutually exclusive with
519 /// [`Self::on_tx`] — if both are set, `on_tx` wins (you're
520 /// already inside a transaction, so wrapping again would deadlock
521 /// or fail on backends without nested transactions).
522 pub fn atomic(mut self) -> Self {
523 self.atomic = Some(true);
524 self
525 }
526
527 /// Opt this QuerySet's write terminal out of the global
528 /// `App::builder().atomic_transactions(true)` default. Useful in
529 /// hot-path batches where the caller already owns the outer
530 /// transaction.
531 pub fn non_atomic(mut self) -> Self {
532 self.atomic = Some(false);
533 self
534 }
535
536 /// Resolve whether this QuerySet should auto-wrap its write
537 /// terminal in a transaction. Per-call override > builder global.
538 pub(crate) fn should_atomic_wrap(&self) -> bool {
539 self.atomic.unwrap_or_else(crate::db::atomic_default)
540 }
541
542 /// Snapshot the rows this queryset selects — the pre-image an audited write
543 /// records (gaps3 #54). Empty (and free) unless the model is `audited`.
544 async fn audit_pre(&self, backend: &str) -> Vec<serde_json::Map<String, JsonValue>>
545 where
546 T: crate::orm::Model,
547 {
548 let meta = crate::migrate::ModelMeta::for_::<T>();
549 if !meta.audited {
550 return Vec::new();
551 }
552 let mut conds: Vec<sea_query::Condition> = Vec::new();
553 for p in &self.predicates {
554 conds.push(sea_query::Condition::all().add(p.cond_for(backend)));
555 }
556 for e in self.implicit_predicates() {
557 conds.push(sea_query::Condition::all().add(e));
558 }
559 crate::orm::dynamic::audit_snapshot(&meta, &conds).await
560 }
561
562 /// Record an audited write against the rows `ids` names.
563 async fn audit_post(
564 before: Vec<serde_json::Map<String, JsonValue>>,
565 ids: &[JsonValue],
566 action: &str,
567 ) where
568 T: crate::orm::Model,
569 {
570 let meta = crate::migrate::ModelMeta::for_::<T>();
571 if !meta.audited {
572 return;
573 }
574 // DELETE has no after-image; for CREATE/UPDATE re-read BY PK — re-running
575 // the caller's filter would miss a row whose filtered column just changed.
576 let after = if action == crate::orm::audit::DELETE {
577 Vec::new()
578 } else {
579 match crate::orm::audit::pk_in_condition(&meta, ids) {
580 Some(c) => crate::orm::dynamic::audit_snapshot(&meta, &[c]).await,
581 None => Vec::new(),
582 }
583 };
584 let pairs = if action == crate::orm::audit::CREATE {
585 after
586 .into_iter()
587 .map(|a| (crate::orm::audit::pk_of(&meta, &a), None, Some(a)))
588 .collect()
589 } else {
590 crate::orm::dynamic::audit_pairs(&meta, before, after)
591 };
592 crate::orm::audit::record_many(&meta, action, pairs).await;
593 }
594
595 /// Predicates the ORM adds on its own, independent of what the caller
596 /// filtered on — today the soft-delete visibility guard.
597 ///
598 /// **This is the single seam for implicit filtering, and it exists so a
599 /// write path cannot silently diverge from the read path.** Both
600 /// [`Self::build_query_for`] (SELECT) and [`Self::build_update_for`]
601 /// (UPDATE) call it, so a row hidden from reads cannot be updated through a
602 /// predicate that forgot the same guard. It was duplicated in both before;
603 /// two copies of a security-relevant filter is a drift bug waiting to
604 /// happen, and the next filter to land here — the row-level tenant scope —
605 /// must reach *every* builder or it is a cross-tenant leak
606 /// (`docs/specs/row-level-tenancy.md`).
607 ///
608 /// Note `build_delete_for` deliberately does NOT call this: a hard delete
609 /// may legitimately target already-trashed rows. A tenant predicate, by
610 /// contrast, will have to apply there too.
611 fn implicit_predicates(&self) -> Vec<sea_query::SimpleExpr> {
612 let mut out = Vec::new();
613 if self.soft_delete_active {
614 use sea_query::Expr;
615 if self.only_deleted {
616 out.push(Expr::col(Alias::new("deleted_at")).is_not_null());
617 } else if !self.with_deleted {
618 out.push(Expr::col(Alias::new("deleted_at")).is_null());
619 }
620 }
621 out
622 }
623
624 /// Clone the base query and weave in the accumulated predicates (plus the
625 /// [`implicit_predicates`](Self::implicit_predicates)), picking the
626 /// dialect-appropriate `SimpleExpr` for each one. The `backend_name` is
627 /// `"sqlite"` or `"postgres"`; any other value behaves like Postgres (the
628 /// default).
629 pub(crate) fn build_query_for(&self, backend_name: &str) -> sea_query::SelectStatement {
630 let mut q = self.query.clone();
631 for p in &self.predicates {
632 q.and_where(p.cond_for(backend_name));
633 }
634 // Feature #72 — soft-delete auto-filter. When the model
635 // opted in via `#[umbral(soft_delete)]` AND the caller
636 // didn't switch the visibility via `.with_deleted()` /
637 // `.only_deleted()`, inject `WHERE deleted_at IS NULL`.
638 // `.with_deleted()` shows everything; `.only_deleted()`
639 // shows just the soft-deleted rows.
640 for implicit in self.implicit_predicates() {
641 q.and_where(implicit);
642 }
643 // Related-aggregate annotations: one correlated scalar
644 // subquery per entry, aliased onto the SELECT list. Living
645 // HERE is what makes `.annotate_*` compose with everything —
646 // explain(), to_sql(), fetch_annotated() all see the same
647 // query. Poisoned entries (unknown relation) are skipped in
648 // this infallible path; the fallible consumers call
649 // `check_annotations()` first and fail loudly instead.
650 for ann in &self.annotations {
651 // M2M-junction annotations count rows of the junction table
652 // (`<parent>_<field>`, columns parent_id / child_id),
653 // correlated on parent_id = <parent>.<pk>.
654 if let Some(junction) = &ann.m2m_junction {
655 if let Ok((_child_table, _fk_col, parent_table, parent_pk)) = &ann.resolved {
656 let mut sub = sea_query::Query::select();
657 sub.expr(ann.agg.to_simple_expr())
658 .from(crate::db::router::schema_qualified_table(junction.as_str()))
659 .and_where(
660 sea_query::Expr::col((
661 Alias::new(junction.as_str()),
662 Alias::new("parent_id"),
663 ))
664 .equals((
665 Alias::new(parent_table.as_str()),
666 Alias::new(parent_pk.as_str()),
667 )),
668 );
669 q.expr_as(
670 sea_query::SimpleExpr::SubQuery(
671 None,
672 Box::new(sea_query::SubQueryStatement::SelectStatement(
673 sub.to_owned(),
674 )),
675 ),
676 Alias::new(ann.alias.as_str()),
677 );
678 }
679 continue;
680 }
681 if let Ok((child_table, fk_col, parent_table, parent_pk)) = &ann.resolved {
682 let mut sub = sea_query::Query::select();
683 sub.expr(ann.agg.to_simple_expr())
684 .from(crate::db::router::schema_qualified_table(
685 child_table.as_str(),
686 ))
687 .and_where(
688 sea_query::Expr::col((
689 Alias::new(child_table.as_str()),
690 Alias::new(fk_col.as_str()),
691 ))
692 .equals((
693 Alias::new(parent_table.as_str()),
694 Alias::new(parent_pk.as_str()),
695 )),
696 );
697 // Auto-exclude soft-deleted children from the count when
698 // the child model is `#[umbral(soft_delete)]`.
699 if ann.child_soft_delete {
700 sub.and_where(
701 sea_query::Expr::col((
702 Alias::new(child_table.as_str()),
703 Alias::new("deleted_at"),
704 ))
705 .is_null(),
706 );
707 }
708 // Child-side predicate (annotate_count_where).
709 if let Some(filter) = &ann.child_filter {
710 sub.and_where(filter.clone());
711 }
712 q.expr_as(
713 sea_query::SimpleExpr::SubQuery(
714 None,
715 Box::new(sea_query::SubQueryStatement::SelectStatement(
716 sub.to_owned(),
717 )),
718 ),
719 Alias::new(ann.alias.as_str()),
720 );
721 }
722 }
723 // gaps3 #29: ORDER BY an annotation alias — "top scorers by goal count".
724 // Both backends allow ordering by a SELECT-list alias, which is what the
725 // annotation is. Applied before the model-default ordering so an explicit
726 // annotation sort wins, exactly like `order_by` does.
727 for (alias, desc) in &self.annotation_order {
728 let order = if *desc { Order::Desc } else { Order::Asc };
729 q.order_by(Alias::new(alias.as_str()), order);
730 }
731 // BUG-8: default ORDER BY applies only when the caller didn't
732 // supply an explicit `.order_by(...)`: the model-default
733 // ordering semantics.
734 if !self.explicit_order {
735 for (col, desc) in &self.default_ordering {
736 let order = if *desc { Order::Desc } else { Order::Asc };
737 q.order_by(Alias::new(*col), order);
738 }
739 }
740 // audit_2 plugin-storage-tasks #6 — `FOR UPDATE SKIP LOCKED`, Postgres
741 // ONLY. It lets concurrent claimers skip rows another txn already locked
742 // (each grabs a different row) instead of all blocking on the head row's
743 // lock. SQLite has no such clause and its single-writer model makes it
744 // unnecessary, so it's a no-op there (never appended).
745 if self.for_update_skip_locked && backend_name != "sqlite" {
746 q.lock_with_behavior(
747 sea_query::LockType::Update,
748 sea_query::LockBehavior::SkipLocked,
749 );
750 }
751 q
752 }
753}
754
755/// Chainable methods on every `QuerySet<T>`.
756///
757/// These are model-agnostic: they only touch the sea-query
758/// `SelectStatement` and the pool-resolution slot, neither of which
759/// depends on `T`. Terminals (which need row mapping) live in the
760/// `impl<T: Model> QuerySet<T>` block below.
761impl<T> QuerySet<T> {
762 /// Add a WHERE condition. Multiple `.filter` calls AND together
763 /// (sea-query's `and_where` semantics — applied at terminal time
764 /// once the resolved pool's backend is known).
765 pub fn filter(mut self, p: Predicate<T>) -> Self {
766 self.predicates.push(p);
767 self
768 }
769
770 /// Add a negated WHERE condition. The negated predicate ANDs into
771 /// the chain alongside any `filter()` calls, so
772 /// `.filter(A).exclude(B).filter(C)` renders as `WHERE A AND NOT B
773 /// AND C`. Sugar for `filter(Q::not(p))`.
774 ///
775 /// The negated-filter terminal.
776 pub fn exclude(self, p: Predicate<T>) -> Self {
777 self.filter(crate::orm::Q::not(p))
778 }
779
780 /// Add an ORDER BY clause. Multiple `.order_by` calls append.
781 /// The first explicit call also opts out of the model's
782 /// `#[umbral(ordering = [...])]` default (BUG-8): explicit ordering
783 /// replaces the default rather than stacking on top of it.
784 pub fn order_by(mut self, o: OrderExpr<T>) -> Self {
785 let order = if o.descending {
786 Order::Desc
787 } else {
788 Order::Asc
789 };
790 self.query.order_by(Alias::new(o.column), order);
791 self.explicit_order = true;
792 self
793 }
794
795 /// Set LIMIT.
796 pub fn limit(mut self, n: u64) -> Self {
797 self.query.limit(n);
798 self.user_limit = Some(n);
799 self
800 }
801
802 /// Set OFFSET.
803 pub fn offset(mut self, n: u64) -> Self {
804 self.query.offset(n);
805 self.user_offset = Some(n);
806 self
807 }
808
809 /// Append `FOR UPDATE SKIP LOCKED` to a read terminal — **Postgres only**
810 /// (a no-op on SQLite). Rows another transaction has already locked are
811 /// skipped rather than blocked on, so N concurrent workers running the same
812 /// `SELECT ... LIMIT k` each claim DIFFERENT rows instead of all contending
813 /// for the same head row. The canonical use is a task/job queue's claim
814 /// query (audit_2 plugin-storage-tasks #6): pair it with a conditional
815 /// `UPDATE ... WHERE status = 'pending'` inside the same transaction.
816 ///
817 /// Must be used inside a transaction (`.on_tx(&mut tx)`) — the row locks a
818 /// bare `SELECT` takes are released immediately, defeating the point.
819 pub fn for_update_skip_locked(mut self) -> Self {
820 self.for_update_skip_locked = true;
821 self
822 }
823
824 /// Override the pool resolved at terminal time with a SQLite pool.
825 ///
826 /// Wins over the ambient default. Used by tests that drive the ORM
827 /// without going through `App::build()`. For a Postgres override
828 /// use [`Self::on_pg`].
829 pub fn on(mut self, pool: &sqlx::SqlitePool) -> Self {
830 self.explicit_pool = Some(DbPool::Sqlite(pool.clone()));
831 self
832 }
833
834 /// Override the pool resolved at terminal time with a Postgres pool.
835 ///
836 /// The Postgres counterpart of [`Self::on`]. Tests that want to
837 /// exercise the Postgres branch (or that drive against a real
838 /// Postgres instance) reach for this directly.
839 pub fn on_pg(mut self, pool: &sqlx::PgPool) -> Self {
840 self.explicit_pool = Some(DbPool::Postgres(pool.clone()));
841 self
842 }
843
844 /// Attach this `QuerySet` to an open transaction.
845 ///
846 /// Returns a [`QuerySetTx`] that holds both the query and a mutable
847 /// reference to the transaction. Every terminal on `QuerySetTx`
848 /// (`fetch`, `first`, `count`, `exists`, `get`, `delete`,
849 /// `update_values`) executes inside the open transaction so all
850 /// operations in the same closure commit or roll back as a unit.
851 ///
852 /// ```rust,ignore
853 /// umbral::db::transaction(|tx| async move {
854 /// let order = Order::objects().on_tx(tx).create(new_order).await?;
855 /// Stock::objects()
856 /// .on_tx(tx)
857 /// .filter(stock::SKU.eq(sku))
858 /// .update_values(delta)
859 /// .await?;
860 /// Ok::<_, MyError>(order)
861 /// }).await?;
862 /// ```
863 pub fn on_tx(self, tx: &mut crate::db::Transaction) -> QuerySetTx<'_, T> {
864 QuerySetTx { qs: self, tx }
865 }
866
867 /// Eagerly load a single FK field by name.
868 ///
869 /// After the main SELECT returns rows, a batch `SELECT ... FROM <related_table>
870 /// WHERE id IN (...)` fetches all referenced rows in one round-trip. Each
871 /// returned row is deserialised as the target model and stored in
872 /// `ForeignKey<U>.resolved` so template rendering (`{{ post.author.username }}`)
873 /// and `serde_json::to_value(&post)["author"]["username"]` both work without
874 /// additional queries.
875 ///
876 /// Calling `select_related` multiple times accumulates the names:
877 /// `.select_related("author").select_related("editor")` works the same as
878 /// `.select_related_many(&["author", "editor"])`.
879 ///
880 /// ## Nested traversal (post-#42)
881 ///
882 /// `.select_related("author__manager")` walks the FK chain through
883 /// the `__` separator. One batched `IN (...)` query per hop —
884 /// `1 + len(hops)` round-trips regardless of parent count. No
885 /// N+1. Each hop's related row is embedded into the prior level's
886 /// JSON, and recursive `ForeignKey<T>::Deserialize` unpacks the
887 /// chain into `resolved()` slots at every depth. Bonus: a
888 /// select_related'd model now round-trips through
889 /// `serde_json::to_value(&t)` / `from_value` without losing the
890 /// resolved relation.
891 ///
892 /// ## Companion shapes
893 ///
894 /// - **`join_related(name)`** — same goal (load related rows) via
895 /// a true `LEFT JOIN` in the main SELECT. One round-trip total
896 /// vs. `select_related`'s `1 + N` batched-IN approach. Wider
897 /// per-row payload; better when round-trip count dominates.
898 /// - **`prefetch_related(name)`** — M2M batched loading (one
899 /// query per declared M2M field). For reverse-FK collections
900 /// (`prefetch_related("comment_set")`-style) see gap #44 — not
901 /// yet implemented.
902 ///
903 /// ## Loud errors
904 ///
905 /// Unknown field names (typos, M2M names accidentally passed
906 /// here, fields without `fk_target`) return a clear
907 /// `sqlx::Error::Protocol` from the terminal naming the bad hop
908 /// and the table it was looked up against. Pre-#42 these
909 /// silently no-op'd.
910 pub fn select_related(mut self, field_name: impl Into<String>) -> Self {
911 self.select_related.push(field_name.into());
912 self
913 }
914
915 /// Eagerly load multiple FK fields in one call.
916 ///
917 /// Sugar for chained `.select_related(name)` calls.
918 pub fn select_related_many(mut self, field_names: &[&str]) -> Self {
919 for name in field_names {
920 self.select_related.push(name.to_string());
921 }
922 self
923 }
924
925 /// JOIN-based eager FK loading — emits `LEFT JOIN <related> ON ...`
926 /// in the main SELECT (with aliased child columns `<field>__<col>`)
927 /// so one round-trip pulls the parent + related rows together.
928 ///
929 /// Trade-off vs. [`Self::select_related`]:
930 /// - `select_related` runs ONE extra batched query after the
931 /// main fetch (`SELECT * FROM related WHERE id IN (...)`).
932 /// Two round-trips total; rows stay narrow.
933 /// - `join_related` runs the main query AS the join — one
934 /// round-trip total — at the cost of a wider per-row payload
935 /// (every related column rides along even for duplicated
936 /// parents).
937 ///
938 /// Both populate `ForeignKey<U>.resolved` the same way so
939 /// downstream code (templates, serde) doesn't care which path
940 /// was used. Pick `join_related` when round-trip count dominates
941 /// (hot listing pages, small related tables) and `select_related`
942 /// when the related row is wide or only loaded for a subset of
943 /// the parent rows.
944 ///
945 /// Composes with `.select_related(other_fk)` and
946 /// `.prefetch_related(m2m)` — different fields can take different
947 /// paths in the same query.
948 ///
949 /// Multi-hop FK chains are supported: a `"__"`-separated path like
950 /// `"author__manager"` resolves one JOIN per hop in a single query.
951 ///
952 /// **Constraints**: FK fields must live in `model.fields` (M2M links
953 /// route through `prefetch_related`), and every related model along the
954 /// chain must be registered with the framework
955 /// (`App::builder().model::<U>()` or contributed by a plugin) so we can
956 /// resolve its column layout for the aliased SELECT.
957 pub fn join_related(mut self, field_name: impl Into<String>) -> Self {
958 self.join_related.push(JoinReq {
959 path: field_name.into(),
960 kind: None,
961 });
962 self
963 }
964
965 /// Sugar for chained [`Self::join_related`] calls.
966 pub fn join_related_many(mut self, field_names: &[&str]) -> Self {
967 for name in field_names {
968 self.join_related.push(JoinReq {
969 path: (*name).to_string(),
970 kind: None,
971 });
972 }
973 self
974 }
975
976 /// `LEFT JOIN` the related path — keeps parent rows whose relation
977 /// is absent (the relation hydrates as unresolved/None). Accepts a
978 /// nested path (`"plugin__author"`); the join type applies to the
979 /// deepest hop.
980 pub fn left_join_related(mut self, path: impl Into<String>) -> Self {
981 self.join_related.push(JoinReq {
982 path: path.into(),
983 kind: Some(JoinKind::Left),
984 });
985 self
986 }
987
988 /// `INNER JOIN` the related path — drops parent rows whose relation
989 /// is absent. The default for a NOT NULL FK.
990 pub fn inner_join_related(mut self, path: impl Into<String>) -> Self {
991 self.join_related.push(JoinReq {
992 path: path.into(),
993 kind: Some(JoinKind::Inner),
994 });
995 self
996 }
997
998 /// `RIGHT JOIN` the related path. Postgres-unconditional; SQLite
999 /// needs >= 3.39 — a runtime warning fires on older SQLite (see the
1000 /// boot/runtime note in the joins docs). The precise version gate
1001 /// lives at execute time (the SQLite driver's own error); the warn
1002 /// is the early nudge.
1003 pub fn right_join_related(mut self, path: impl Into<String>) -> Self {
1004 self.join_related.push(JoinReq {
1005 path: path.into(),
1006 kind: Some(JoinKind::Right),
1007 });
1008 self
1009 }
1010
1011 /// Eagerly load an M2M relation via a single batched join.
1012 ///
1013 /// After the main SELECT returns rows, one query of the shape
1014 /// `SELECT j.parent_id, child.* FROM <child_table> child INNER
1015 /// JOIN <junction> j ON child.<pk> = j.child_id WHERE
1016 /// j.parent_id IN (...)` fetches every related child for every
1017 /// parent in one round-trip. Each parent's `M2M.resolved` slot
1018 /// is populated with its matching children.
1019 ///
1020 /// The M2M counterpart of [`Self::select_related`] for FKs —
1021 /// same goal of killing N+1: a batch-loaded `prefetch_related('tags')`.
1022 ///
1023 /// ## Reverse-FK collections (post-#44)
1024 ///
1025 /// `prefetch_related` also loads `ReverseSet<C>` fields — the
1026 /// "for each Post, give me every Comment that points at it"
1027 /// shape. Declare the field on the parent with
1028 /// `#[sqlx(skip)] #[serde(skip)]
1029 /// #[umbral(reverse_fk = "<fk_col>")] pub <name>: ReverseSet<C>`
1030 /// where `<fk_col>` names the FK column on `C` pointing back.
1031 /// One `SELECT * FROM <child> WHERE <fk_col> IN (parent_pks)`
1032 /// regardless of parent count — no N+1.
1033 ///
1034 /// ## Scope (v1)
1035 ///
1036 /// - **M2M + reverse-FK only.** FK fields go through
1037 /// [`Self::select_related`] (batched IN) or
1038 /// [`Self::join_related`] (LEFT JOIN).
1039 /// - **i64 parent PK only.** Same constraint as the rest of the
1040 /// M2M plumbing; models with non-i64 PKs surface a clean
1041 /// compile error.
1042 /// - **Unknown field name → loud error** (post-#42). If the
1043 /// name matches neither an M2M nor a `ReverseSet` field,
1044 /// fetch returns a clear `sqlx::Error::Protocol` pointing at
1045 /// the right method.
1046 pub fn prefetch_related(mut self, field_name: impl Into<String>) -> Self {
1047 self.prefetch_related.push(field_name.into());
1048 self
1049 }
1050
1051 /// Eagerly load multiple M2M relations. Sugar for chained
1052 /// `.prefetch_related(name)` calls.
1053 pub fn prefetch_related_many(mut self, field_names: &[&str]) -> Self {
1054 for name in field_names {
1055 self.prefetch_related.push(name.to_string());
1056 }
1057 self
1058 }
1059
1060 /// Convert this QuerySet into a [`Subquery`] suitable for use in
1061 /// an `IN (SELECT ...)` predicate. Projects only the named
1062 /// column; the accumulated WHERE / ORDER BY survive.
1063 ///
1064 /// `Post::objects().filter(...).into_subquery("author_id")` →
1065 /// `Subquery` you can hand to `user::ID.in_subquery(...)`.
1066 pub fn into_subquery(self, col_name: &str) -> crate::orm::Subquery {
1067 // Resolve the real backend so per-backend predicates (e.g. a JSON
1068 // path's `json_extract(...)` vs `col ->> 'k'`) render for the pool
1069 // that will run the OUTER query — hardcoding "sqlite" baked the
1070 // SQLite variant into subqueries later executed on Postgres.
1071 let backend = subquery_backend(&self.explicit_pool);
1072 let mut q = self.build_query_for(backend);
1073 q.clear_selects();
1074 q.column(Alias::new(col_name));
1075 crate::orm::Subquery::from_select(q)
1076 }
1077
1078 /// Combine this QuerySet with `other` via SQL `UNION` (gap #28).
1079 /// Both QuerySets must produce the same column shape — which
1080 /// they always do here because both are typed `QuerySet<T>`.
1081 /// Duplicates are removed (the de-duplicating UNION, not UNION
1082 /// ALL).
1083 pub fn union(self, other: QuerySet<T>) -> Self {
1084 self.combine(other, sea_query::UnionType::Distinct)
1085 }
1086
1087 /// Combine this QuerySet with `other` via SQL `INTERSECT`
1088 /// (gap #28). Returns rows present in BOTH inputs.
1089 pub fn intersect(self, other: QuerySet<T>) -> Self {
1090 self.combine(other, sea_query::UnionType::Intersect)
1091 }
1092
1093 /// Combine this QuerySet with `other` via SQL `EXCEPT`
1094 /// (gap #28). Returns rows present in `self` but not in `other`.
1095 pub fn except(self, other: QuerySet<T>) -> Self {
1096 self.combine(other, sea_query::UnionType::Except)
1097 }
1098
1099 /// Internal: attach `other`'s SelectStatement to `self`'s
1100 /// SelectStatement with the given UnionType. Both sides apply
1101 /// their accumulated predicates / ORDER BY before the union.
1102 fn combine(mut self, other: QuerySet<T>, ty: sea_query::UnionType) -> Self {
1103 // Resolve the real backend (not a hardcoded "sqlite") so a JSON-path
1104 // predicate on either arm renders for the pool that runs the
1105 // combined query — see `into_subquery`.
1106 let backend = subquery_backend(&self.explicit_pool);
1107 let other_select = other.build_query_for(backend);
1108 // Fold our own predicates into the base query so the union
1109 // sees them; further `.filter()` calls on the returned
1110 // QuerySet would still apply to the OUTER (combined) query.
1111 let mut base = self.build_query_for(backend);
1112 self.predicates.clear();
1113 base.union(ty, other_select);
1114 self.query = base;
1115 self
1116 }
1117
1118 /// Emit `SELECT DISTINCT ...` for this query (gap #17). Most
1119 /// useful when combined with [`Self::values`] to dedupe a
1120 /// column-projected list (`distinct().values(&["tag"])`); the
1121 /// full-row DISTINCT is rarely what you want.
1122 ///
1123 /// Postgres-specific `DISTINCT ON (cols)` is deferred until a
1124 /// real consumer surfaces the need — the standard `DISTINCT`
1125 /// covers most use cases.
1126 pub fn distinct(mut self) -> Self {
1127 self.query.distinct();
1128 self
1129 }
1130}
1131
1132/// Resolve the pool to run a terminal against.
1133///
1134/// Precedence: explicit `.on(&pool)` / `.on_pg(&pool)` override wins;
1135/// then the per-model database alias the Plugin contract published
1136/// via `Plugin::database()` (FEATURES.md #6); then the `"default"`
1137/// pool. Tests that skip the App builder pass an explicit pool and
1138/// bypass the alias lookup entirely.
1139/// Validate that every name passed to `.join_related(...)` resolves
1140/// to a foreign-key field on `T`. Loud-error path that replaces the
1141/// pre-#42 silent no-op (where an unknown field, an M2M field, or a
1142/// non-FK column produced an empty JOIN list and a confusing
1143/// "ForeignKey::resolved() returned None" downstream).
1144fn validate_join_related_fields<T: Model>(fields: &[String]) -> Result<(), sqlx::Error> {
1145 for field_name in fields {
1146 // Nested path (`"plugin__author"` / `"tags__category"`):
1147 // validate via the hop resolver. A leading M2M segment is a
1148 // valid chain entry even though it isn't an FK on `T`, so
1149 // accept it and let `apply_join_related` route it.
1150 if field_name.contains("__") {
1151 let first = field_name.split("__").next().unwrap_or(field_name);
1152 let leads_m2m = T::M2M_RELATIONS.iter().any(|r| r.field_name == first);
1153 if leads_m2m || resolve_join_hops::<T>(field_name).is_some() {
1154 continue;
1155 }
1156 return Err(sqlx::Error::Protocol(format!(
1157 "umbral::orm::join_related: nested path `{field_name}` on `{}` has an \
1158 unresolvable hop (each segment must be a FK or M2M to a registered model)",
1159 T::NAME
1160 )));
1161 }
1162 // Try as a regular column first.
1163 let col = T::FIELDS.iter().find(|f| f.name == field_name.as_str());
1164 if let Some(col) = col {
1165 if col.fk_target.is_some() {
1166 continue; // FK column — OK.
1167 }
1168 return Err(sqlx::Error::Protocol(format!(
1169 "umbral::orm::join_related: field `{field_name}` on `{}` is not a foreign \
1170 key (it has no fk_target)",
1171 T::NAME
1172 )));
1173 }
1174 // M2M field name? Post-#113 these go through the double
1175 // LEFT JOIN path: apply_join_related emits
1176 // `LEFT JOIN <junction> LEFT JOIN <child>` with aliased
1177 // child cols, and fetch()'s dedup-aware path collects M2M
1178 // children per parent. Trade-off documented at the join_related
1179 // docstring — M2M JOINs multiply parent rows by avg
1180 // cardinality, so prefetch_related stays the default for
1181 // any list page where M2M cardinality isn't tiny.
1182 if T::M2M_RELATIONS
1183 .iter()
1184 .any(|r| r.field_name == field_name.as_str())
1185 {
1186 continue;
1187 }
1188 return Err(sqlx::Error::Protocol(format!(
1189 "umbral::orm::join_related: unknown field `{field_name}` on model `{}`",
1190 T::NAME
1191 )));
1192 }
1193 Ok(())
1194}
1195
1196/// One resolved hop of a `join_related` FK chain.
1197#[derive(Debug, Clone)]
1198pub(crate) struct JoinHop {
1199 /// FK column name on the *previous* level's table.
1200 pub(crate) fk_col: String,
1201 /// Table this hop targets.
1202 pub(crate) child_table: String,
1203 /// PK column on `child_table`.
1204 pub(crate) child_pk: String,
1205 /// Was the FK column nullable? (drives auto-inference)
1206 pub(crate) nullable: bool,
1207}
1208
1209/// Resolve a dotted FK path (`"plugin__author"`) into ordered hops.
1210/// Hop 0 reads `T::FIELDS`; deeper hops read the migrate registry's
1211/// `Column`s for the prior hop's target table. Returns `None` (skip,
1212/// emit no JOIN) on any unresolved hop — same forgiving posture as the
1213/// pre-existing one-hop path's silent skip in `to_sql`.
1214///
1215/// FK-only: a path whose FIRST segment is an M2M field is NOT handled
1216/// here (M2M chains route through `apply_join_related`'s M2M branch);
1217/// this returns `None` for such a path.
1218pub(crate) fn resolve_join_hops<T: Model>(path: &str) -> Option<Vec<JoinHop>> {
1219 let registered = crate::migrate::registered_models();
1220 let segs: Vec<&str> = path.split("__").filter(|s| !s.is_empty()).collect();
1221 if segs.is_empty() {
1222 return None;
1223 }
1224 let mut hops = Vec::with_capacity(segs.len());
1225 // Hop 0 off the typed parent.
1226 let f0 = T::FIELDS.iter().find(|f| f.name == segs[0])?;
1227 let t0 = f0.fk_target?;
1228 let m0 = registered.iter().find(|m| m.table == t0)?;
1229 let pk0 = m0.fields.iter().find(|c| c.primary_key)?;
1230 hops.push(JoinHop {
1231 fk_col: segs[0].to_string(),
1232 child_table: t0.to_string(),
1233 child_pk: pk0.name.clone(),
1234 nullable: f0.nullable,
1235 });
1236 let mut current = t0;
1237 for seg in &segs[1..] {
1238 let meta = registered.iter().find(|m| m.table == current)?;
1239 let col = meta.fields.iter().find(|c| c.name == *seg)?;
1240 let tgt = col.fk_target.as_deref()?;
1241 let tmeta = registered.iter().find(|m| m.table == tgt)?;
1242 let pk = tmeta.fields.iter().find(|c| c.primary_key)?;
1243 hops.push(JoinHop {
1244 fk_col: (*seg).to_string(),
1245 child_table: tgt.to_string(),
1246 child_pk: pk.name.clone(),
1247 nullable: col.nullable,
1248 });
1249 current = tgt;
1250 }
1251 Some(hops)
1252}
1253
1254/// Thin wrapper so the backend hydration helpers (a sibling module)
1255/// can resolve a chain without importing the private name directly.
1256pub(crate) fn resolve_join_hops_for<T: Model>(path: &str) -> Option<Vec<JoinHop>> {
1257 resolve_join_hops::<T>(path)
1258}
1259
1260/// Resolve a path whose FIRST segment is an M2M field on `T` into the
1261/// M2M child table + child PK + the onward FK chain hops off that
1262/// child. `onward` is empty for a bare `"tags"`; for `"tags__category"`
1263/// it carries the `category` FK hop off the child table. `None` when
1264/// `segs[0]` isn't an M2M field or any onward hop fails to resolve.
1265pub(crate) fn resolve_m2m_chain<T: Model>(path: &str) -> Option<(String, String, Vec<JoinHop>)> {
1266 let registered = crate::migrate::registered_models();
1267 let segs: Vec<&str> = path.split("__").filter(|s| !s.is_empty()).collect();
1268 let first = segs.first()?;
1269 let rel = T::M2M_RELATIONS.iter().find(|r| r.field_name == *first)?;
1270 let child_meta = registered.iter().find(|m| m.table == rel.target_table)?;
1271 let child_pk = child_meta.fields.iter().find(|c| c.primary_key)?;
1272 let mut onward = Vec::with_capacity(segs.len().saturating_sub(1));
1273 let mut current = rel.target_table;
1274 for seg in &segs[1..] {
1275 let meta = registered.iter().find(|m| m.table == current)?;
1276 let col = meta.fields.iter().find(|c| c.name == *seg)?;
1277 let tgt = col.fk_target.as_deref()?;
1278 let tmeta = registered.iter().find(|m| m.table == tgt)?;
1279 let pk = tmeta.fields.iter().find(|c| c.primary_key)?;
1280 onward.push(JoinHop {
1281 fk_col: (*seg).to_string(),
1282 child_table: tgt.to_string(),
1283 child_pk: pk.name.clone(),
1284 nullable: col.nullable,
1285 });
1286 current = tgt;
1287 }
1288 Some((rel.target_table.to_string(), child_pk.name.clone(), onward))
1289}
1290
1291/// Gap #111 — error returned when a typed terminal (`fetch` / `first`
1292/// / `get`) runs against a QuerySet that has `.only(...)` set. A
1293/// partial-column row can't satisfy `T`'s `FromRow` impl, so the
1294/// caller has to either drop the `.only(...)` (full SELECT, typed
1295/// rows back) or terminate via `.values(&[...])` (JSON rows with
1296/// just the requested columns). The message names the offending
1297/// terminal so the fix is one rename away.
1298fn only_with_typed_terminal_error(terminal: &'static str) -> sqlx::Error {
1299 sqlx::Error::Protocol(format!(
1300 "umbral::orm::{terminal}: cannot run a typed terminal on a QuerySet \
1301 with `.only(...)` set — a partial-column row can't hydrate `T` via \
1302 FromRow. Either drop `.only(...)` to fetch full typed rows, or \
1303 terminate via `.values(&[...])` to get JSON rows with just the \
1304 projected columns."
1305 ))
1306}
1307
1308/// gaps4 #24: the `_pg` explicit-pool terminals are LOW-LEVEL — they build and
1309/// run the base query and nothing else. Unlike the generic `fetch` / `first` /
1310/// `get`, they do NOT apply `select_related` / `prefetch_related` /
1311/// `join_related` hydration, `.only()` validation, or M2M parent-id seeding.
1312///
1313/// Silently ignoring a chained `.select_related(...)` and handing back
1314/// un-hydrated rows is the worst outcome — the caller debugs the wrong thing.
1315/// So a `_pg` terminal REFUSES when one of those features is set, naming it and
1316/// pointing at the generic terminal (which routes to the right backend via the
1317/// ambient pool / `DatabaseRouter` anyway).
1318fn guard_pg_terminal_unsupported(
1319 terminal: &'static str,
1320 select_related: &[String],
1321 prefetch_related: &[String],
1322 join_related: &[JoinReq],
1323 only_cols: &Option<Vec<String>>,
1324) -> Result<(), sqlx::Error> {
1325 let feature = if only_cols.is_some() {
1326 // `.only()` already has its own dedicated error on the generic path;
1327 // reuse it so the two terminals give the same message.
1328 return Err(only_with_typed_terminal_error(terminal));
1329 } else if !select_related.is_empty() {
1330 "select_related"
1331 } else if !prefetch_related.is_empty() {
1332 "prefetch_related"
1333 } else if !join_related.is_empty() {
1334 "join_related"
1335 } else {
1336 return Ok(());
1337 };
1338 Err(sqlx::Error::Protocol(format!(
1339 "umbral::orm::{terminal}: `.{feature}(...)` is not applied by the low-level \
1340 `_pg` terminals — they run the base query only, with no relation hydration. \
1341 Silently returning un-hydrated rows would be worse, so this is refused. Use \
1342 the generic `.{}(...)` terminal instead (it routes to Postgres through the \
1343 ambient pool / DatabaseRouter and DOES hydrate), or drop `.{feature}(...)`.",
1344 terminal.trim_end_matches("_pg"),
1345 )))
1346}
1347
1348/// Resolve the backend name for a pure query-BUILDING context (subquery /
1349/// union composition), where there is no executor yet. Prefer the
1350/// QuerySet's explicit pool, else the ambient default pool; fall back to
1351/// `"sqlite"` when no app is booted (a to_sql-style call in a test) so the
1352/// builder never panics.
1353fn subquery_backend(explicit: &Option<DbPool>) -> &'static str {
1354 if let Some(p) = explicit {
1355 return p.backend_name();
1356 }
1357 crate::db::try_pool_dispatched()
1358 .map(|p| p.backend_name())
1359 .unwrap_or("sqlite")
1360}
1361
1362fn resolve_pool<T: Model>(explicit: Option<DbPool>, op: crate::db::RouteOp) -> DbPool {
1363 if let Some(pool) = explicit {
1364 return pool;
1365 }
1366 // Route through the swappable router when the registry is up.
1367 if let Some(meta) = crate::migrate::model_meta_ref(T::NAME) {
1368 let ctx = crate::db::route_context::current();
1369 let r = crate::db::router::router();
1370 let alias = match op {
1371 crate::db::RouteOp::Read => r.db_for_read(meta, &ctx),
1372 crate::db::RouteOp::Write => r.db_for_write(meta, &ctx),
1373 };
1374 return crate::db::pool_for_dispatched(alias.as_str()).clone();
1375 }
1376 // Registry-less fallback (low-level tests): today's static behavior.
1377 if let Some(alias) = crate::migrate::model_alias(T::NAME) {
1378 return crate::db::pool_for_dispatched(&alias).clone();
1379 }
1380 crate::db::pool_dispatched().clone()
1381}
1382
1383/// Pin a QuerySet to an explicit pool, dispatching SQLite vs Postgres. Used by
1384/// the upsert paths (`get_or_create` / `update_or_create`) so their
1385/// existence-check reads run on the WRITE database — read-your-writes, so a
1386/// read/write-split router never probes a lagging replica and inserts a
1387/// duplicate (or reads a stale row back after the update).
1388fn pin_to_pool<T: Model>(qs: QuerySet<T>, pool: &DbPool) -> QuerySet<T> {
1389 match pool {
1390 DbPool::Sqlite(p) => qs.on(p),
1391 DbPool::Postgres(p) => qs.on_pg(p),
1392 }
1393}
1394
1395// GetError / TryForEachError moved to `errors`; re-exported above.
1396
1397/// Emit a one-shot advisory when a `right_join_related` is applied
1398/// against a SQLite pool.
1399///
1400/// RIGHT/FULL JOIN landed in SQLite 3.39 (June 2022); Postgres has
1401/// always supported it. The boot system check (`check.rs`) can't surface
1402/// this — it's synchronous, has no live pool, and whether a RIGHT join
1403/// is *reachable* is a runtime QuerySet fact rather than static model
1404/// metadata. So the spec's "boot warning" is realized here: the first
1405/// time a RIGHT join is built against a SQLite pool, we `tracing::warn!`
1406/// once per process. We do NOT probe the library version (that needs an
1407/// async round-trip the SQL builder doesn't have) — the precise gate is
1408/// the SQLite driver's own error at execute time on an engine < 3.39;
1409/// this warn is the early nudge, consistent with `check.rs`'s
1410/// `Severity::Warning` posture.
1411///
1412/// Postgres pools and the no-pool case (a pure `to_sql` build with no
1413/// app booted) are silent.
1414fn warn_right_join_on_sqlite() {
1415 use std::sync::Once;
1416 static ONCE: Once = Once::new();
1417 if matches!(
1418 crate::db::try_pool_dispatched(),
1419 Some(crate::db::DbPool::Sqlite(_))
1420 ) {
1421 ONCE.call_once(|| {
1422 tracing::warn!(
1423 "umbral::orm::right_join_related: RIGHT JOIN requires SQLite >= 3.39. \
1424 If your SQLite is older the query will error at execute time; \
1425 Postgres is unaffected. Prefer left_/inner_join_related on SQLite \
1426 unless you've confirmed the engine version."
1427 );
1428 });
1429 }
1430}
1431
1432/// Terminal methods for every `QuerySet<T>` where `T: Model`.
1433///
1434/// Each terminal that materializes `T` carries a FromRow bound on the
1435/// method (not the impl block) — the conjunction of both backends'
1436/// FromRow impls. `#[derive(sqlx::FromRow)]` emits a generic-over-`R`
1437/// impl, so any user struct with standard field types satisfies both
1438/// bounds automatically.
1439impl<T: Model> QuerySet<T> {
1440 /// Render the SQL the QuerySet would execute, without running it.
1441 ///
1442 /// Returns the prepared statement with `?` placeholders for the
1443 /// bound values, exactly the string sqlx would send. Useful for
1444 /// `eprintln!`-style debugging and for tests that want to pin
1445 /// the rendered query without round-tripping through a pool.
1446 ///
1447 /// The bound values are intentionally not surfaced (sqlx's binder
1448 /// types aren't part of umbral's public surface); a `(sql, values)`
1449 /// accessor lands when EXPLAIN-style integration needs it.
1450 ///
1451 /// The rendered placeholder dialect is SQLite's (`?`). When the
1452 /// dispatched pool is Postgres the actual at-execute rendering
1453 /// uses `$1`-style placeholders; the `to_sql` debug surface
1454 /// continues to emit SQLite-style for stability across calls
1455 /// regardless of which pool is registered.
1456 pub fn to_sql(&self) -> String {
1457 let mut q = self.build_query_for("sqlite");
1458 self.apply_join_related(&mut q);
1459 self.apply_only_projection(&mut q);
1460 let (sql, _values) = q.build_sqlx(SqliteQueryBuilder);
1461 sql
1462 }
1463
1464 /// Render the QuerySet's SQL against the **Postgres** dialect,
1465 /// without running it. Companion to [`Self::to_sql`].
1466 ///
1467 /// The two render slightly different placeholder syntax (`?` for
1468 /// SQLite, `$1..$N` for Postgres) and any Postgres-specific
1469 /// operators like the array `@>` / `<@` / `&&` family only render
1470 /// correctly through this entry point — `to_sql`'s SQLite path
1471 /// leaves `$N` tokens in the template untouched. Use this when
1472 /// debugging a Postgres query or asserting on the rendered shape
1473 /// in tests.
1474 pub fn to_sql_pg(&self) -> String {
1475 let mut q = self.build_query_for("postgres");
1476 self.apply_join_related(&mut q);
1477 self.apply_only_projection(&mut q);
1478 let (sql, _values) = q.build_sqlx(PostgresQueryBuilder);
1479 sql
1480 }
1481
1482 /// Internal helper — when `.only(...)` was set, swap the SELECT
1483 /// list for just those columns. Shared by `to_sql` / `to_sql_pg`
1484 /// so the inspection surface stays in sync with what `values()`
1485 /// would emit. No-op when `only_cols` is `None`.
1486 fn apply_only_projection(&self, q: &mut sea_query::SelectStatement) {
1487 if let Some(cols) = &self.only_cols {
1488 q.clear_selects();
1489 for c in cols {
1490 q.column(Alias::new(c.as_str()));
1491 }
1492 }
1493 }
1494
1495 /// Internal helper — when `.join_related(name)` was set, wrap the
1496 /// current query as a subquery and build an outer SELECT that
1497 /// LEFT JOINs every requested related table (with child columns
1498 /// aliased as `<field>__<col>`). The subquery wrapper is the
1499 /// load-bearing trick: WHERE / ORDER BY / LIMIT predicates inside
1500 /// the inner query reference parent columns only — there's no
1501 /// JOIN in scope, so bare names like `id` resolve unambiguously
1502 /// to the parent. Without this wrapper SQLite raises
1503 /// "ambiguous column name: id" on any predicate sharing a column
1504 /// name with a JOIN'd table.
1505 ///
1506 /// No-op when `join_related` is empty. Unknown field names /
1507 /// unregistered related models / FK columns missing `fk_target`
1508 /// are silently skipped — the SQL just won't carry the JOIN and
1509 /// the caller notices when `ForeignKey::resolved()` stays empty.
1510 fn apply_join_related(&self, q: &mut sea_query::SelectStatement) {
1511 if self.join_related.is_empty() {
1512 return;
1513 }
1514 use sea_query::{Expr, Query};
1515 let registered = crate::migrate::registered_models();
1516
1517 // Inner-subquery column trim. When `.only(...)` is also set,
1518 // the outer SELECT only references a subset of parent
1519 // columns (plus the JOIN'd child columns it gets through
1520 // its alias). The inner subquery only needs to expose:
1521 // - parent columns named in `.only(...)` (intersected
1522 // with T::FIELDS so the JOIN aliases like
1523 // `category__name` don't leak in here — those live on
1524 // the JOIN'd table, not on the parent),
1525 // - PLUS the FK columns each `.join_related(name)` needs
1526 // for its `ON __p.<name> = <child>.<pk>` clause.
1527 // WHERE / ORDER BY inside the inner subquery can still
1528 // reference any parent column (SQL doesn't require an
1529 // ORDER BY column to be in the SELECT list), so we don't
1530 // need to promote those. Postgres often skips this prune
1531 // through subquery boundaries; SQLite usually doesn't —
1532 // either way trimming here is a measurable win on wide
1533 // tables (think 30-column Product on a busy hot path).
1534 if let Some(only) = &self.only_cols {
1535 let parent_field_names: std::collections::HashSet<&str> =
1536 T::FIELDS.iter().map(|f| f.name).collect();
1537 let mut needed: std::collections::HashSet<String> = only
1538 .iter()
1539 .filter(|c| parent_field_names.contains(c.as_str()))
1540 .cloned()
1541 .collect();
1542 for jr in &self.join_related {
1543 // Nested paths only need the FIRST hop's FK column at
1544 // the parent level; deeper hops join off the prior
1545 // level's alias, not the parent subquery.
1546 let join_field = jr.path.split("__").next().unwrap_or(jr.path.as_str());
1547 if parent_field_names.contains(join_field) {
1548 needed.insert(join_field.to_string());
1549 }
1550 }
1551 if !needed.is_empty() {
1552 q.clear_selects();
1553 // Stable ordering so the SQL is deterministic
1554 // across runs (HashSet iteration is not).
1555 let mut ordered: Vec<String> = needed.into_iter().collect();
1556 ordered.sort();
1557 for col in &ordered {
1558 q.column(Alias::new(col.as_str()));
1559 }
1560 }
1561 }
1562
1563 // Take ownership of the (possibly-trimmed) inner query and
1564 // re-mount it as the FROM clause of the new outer SELECT.
1565 let inner = std::mem::replace(q, Query::select().take());
1566 let parent_alias = Alias::new("__p");
1567 let mut outer = Query::select();
1568 outer.from_subquery(inner, parent_alias.clone());
1569 // Re-project the parent's full column set so the outer SELECT
1570 // exposes them unaliased — FromRow on `T` reads parent
1571 // columns by their bare names (`id`, `name`, ...). When
1572 // `.only(...)` later clears this list in
1573 // `apply_only_projection`, the inner-subquery trim above
1574 // means we still didn't pay for columns we ended up
1575 // dropping anyway.
1576 for f in T::FIELDS {
1577 outer.expr(Expr::col((parent_alias.clone(), Alias::new(f.name))));
1578 }
1579 // Set when any emitted hop is a RIGHT JOIN — drives the
1580 // once-per-process old-SQLite advisory after the emit loop.
1581 let mut emitted_right = false;
1582 for jr in &self.join_related {
1583 let field_name = &jr.path;
1584 // FK chain branch first. A (possibly nested) FK path splits
1585 // on `__` into ordered hops; each hop joins onto the prior
1586 // level's alias, and the DEEPEST hop's child columns are
1587 // aliased by the full dotted path so hydration can rebuild
1588 // the nested relation graph. The single-hop case is
1589 // byte-identical in child-column aliases to the pre-nesting
1590 // path (`<field>__<col>`); only the internal join alias
1591 // gains an `_h{idx}` suffix, which no test asserts.
1592 if let Some(hops) = resolve_join_hops::<T>(field_name) {
1593 let mut prev_alias = parent_alias.clone();
1594 let last = hops.len() - 1;
1595 // Cumulative dotted prefix per hop so EVERY level's own
1596 // columns ride along, aliased by its path-so-far. Hop 0
1597 // of `plugin__author` is `plugin`, hop 1 is
1598 // `plugin__author`. Selecting every level (not just the
1599 // leaf) is what lets hydration rebuild a FULL nested
1600 // object — the intermediate `plugin` row needs its own
1601 // `id`/`name` to deserialise into `ForeignKey<Plugin>`
1602 // before `author` nests inside it.
1603 let segs: Vec<&str> = field_name.split("__").collect();
1604 for (idx, hop) in hops.iter().enumerate() {
1605 let hop_alias = Alias::new(format!("__j_{field_name}_h{idx}"));
1606 // Last hop: explicit request, else infer from THIS
1607 // hop's nullability. Intermediate hops always infer
1608 // per-hop (an INNER can nest inside an outer
1609 // LEFT etc.); an explicit kind only pins the leaf.
1610 let kind = if idx == last {
1611 jr.kind.unwrap_or(if hop.nullable {
1612 JoinKind::Left
1613 } else {
1614 JoinKind::Inner
1615 })
1616 } else if hop.nullable {
1617 JoinKind::Left
1618 } else {
1619 JoinKind::Inner
1620 };
1621 emitted_right |= kind == JoinKind::Right;
1622 outer.join_as(
1623 kind.sea(),
1624 crate::db::router::schema_qualified_table(hop.child_table.as_str()),
1625 hop_alias.clone(),
1626 Expr::col((prev_alias.clone(), Alias::new(hop.fk_col.as_str())))
1627 .equals((hop_alias.clone(), Alias::new(hop.child_pk.as_str()))),
1628 );
1629 if let Some(meta) = registered.iter().find(|m| m.table == hop.child_table) {
1630 // Cumulative dotted prefix for this hop's columns.
1631 let prefix = segs[..=idx].join("__");
1632 for col in &meta.fields {
1633 let alias = format!("{}__{}", prefix, col.name);
1634 outer.expr_as(
1635 Expr::col((hop_alias.clone(), Alias::new(col.name.as_str()))),
1636 Alias::new(alias),
1637 );
1638 }
1639 }
1640 prev_alias = hop_alias;
1641 }
1642 continue;
1643 }
1644
1645 // M2M branch (post-#113). Emit the double LEFT JOIN
1646 // through the junction table:
1647 // LEFT JOIN <junction> AS __jm_<field>
1648 // ON __p.<parent_pk> = __jm_<field>.parent_id
1649 // LEFT JOIN <child_table> AS __j_<field>
1650 // ON __jm_<field>.child_id = __j_<field>.<child_pk>
1651 // Aliased child cols use the same `<field>__<col>` shape
1652 // as the FK branch so the decode helper can be reused.
1653 // The M2M field is the FIRST segment of the path; a nested
1654 // path like `"tags__category"` passes THROUGH the M2M hop
1655 // and continues with an onward FK chain off the child.
1656 let m2m_seg = field_name.split("__").next().unwrap_or(field_name.as_str());
1657 if let Some(m2m_rel) = T::M2M_RELATIONS.iter().find(|r| r.field_name == m2m_seg)
1658 && let Some(parent_pk) = T::FIELDS.iter().find(|f| f.primary_key)
1659 && let Some(child_meta) =
1660 registered.iter().find(|m| m.table == m2m_rel.target_table)
1661 && let Some(child_pk) = child_meta.fields.iter().find(|c| c.primary_key)
1662 {
1663 // Junction table + aliases key off the M2M field name
1664 // (segs[0]), NOT the full dotted path.
1665 let junction_table = format!("{}_{}", T::TABLE, m2m_seg);
1666 let junction_alias = Alias::new(format!("__jm_{m2m_seg}"));
1667 let child_alias = Alias::new(format!("__j_{m2m_seg}"));
1668 // The junction hop stays LEFT so a parent with zero
1669 // junction rows isn't dropped by the join to the
1670 // junction table itself — the CHILD hop's kind is what
1671 // decides drop/keep. Plain `join_related` (kind None)
1672 // leaves the child LEFT too, preserving the shipped
1673 // double-LEFT-JOIN M2M behavior (a tag-less parent
1674 // survives with an empty M2M slot). An explicit
1675 // inner_join_related drops parents whose relation is
1676 // absent: the junction-LEFT miss yields a NULL child_id,
1677 // then the child INNER on NULL has no match -> the parent
1678 // is dropped, which is the INNER contract.
1679 let child_kind = jr.kind.unwrap_or(JoinKind::Left);
1680 emitted_right |= child_kind == JoinKind::Right;
1681 outer.join_as(
1682 sea_query::JoinType::LeftJoin,
1683 crate::db::router::schema_qualified_table(&junction_table),
1684 junction_alias.clone(),
1685 Expr::col((parent_alias.clone(), Alias::new(parent_pk.name)))
1686 .equals((junction_alias.clone(), Alias::new("parent_id"))),
1687 );
1688 outer.join_as(
1689 child_kind.sea(),
1690 crate::db::router::schema_qualified_table(m2m_rel.target_table),
1691 child_alias.clone(),
1692 Expr::col((junction_alias.clone(), Alias::new("child_id")))
1693 .equals((child_alias.clone(), Alias::new(child_pk.name.as_str()))),
1694 );
1695 // Child columns aliased by the M2M field name so the
1696 // M2M decode path (`<m2m_field>__<col>`) reads them.
1697 for col in &child_meta.fields {
1698 let alias = format!("{}__{}", m2m_seg, col.name);
1699 outer.expr_as(
1700 Expr::col((child_alias.clone(), Alias::new(col.name.as_str()))),
1701 Alias::new(alias),
1702 );
1703 }
1704 // Onward FK chain off the child (segs[1..]). Each hop
1705 // joins onto the prior level's alias and aliases its
1706 // columns by the cumulative dotted path
1707 // (`tags__category__name`) so the M2M decode path can
1708 // nest the onward object into each child row.
1709 if let Some((_child_table, _child_pk, onward)) = resolve_m2m_chain::<T>(field_name)
1710 {
1711 let segs: Vec<&str> = field_name.split("__").collect();
1712 let mut prev_alias = child_alias.clone();
1713 for (i, hop) in onward.iter().enumerate() {
1714 // segs index for this hop: segs[0] is the M2M
1715 // field, segs[1] is onward[0], etc.
1716 let seg_idx = i + 1;
1717 let hop_alias = Alias::new(format!("__j_{m2m_seg}_o{i}"));
1718 let kind = if hop.nullable {
1719 JoinKind::Left
1720 } else {
1721 JoinKind::Inner
1722 };
1723 outer.join_as(
1724 kind.sea(),
1725 crate::db::router::schema_qualified_table(hop.child_table.as_str()),
1726 hop_alias.clone(),
1727 Expr::col((prev_alias.clone(), Alias::new(hop.fk_col.as_str())))
1728 .equals((hop_alias.clone(), Alias::new(hop.child_pk.as_str()))),
1729 );
1730 if let Some(meta) = registered.iter().find(|m| m.table == hop.child_table) {
1731 let prefix = segs[..=seg_idx].join("__");
1732 for col in &meta.fields {
1733 let alias = format!("{}__{}", prefix, col.name);
1734 outer.expr_as(
1735 Expr::col((hop_alias.clone(), Alias::new(col.name.as_str()))),
1736 Alias::new(alias),
1737 );
1738 }
1739 }
1740 prev_alias = hop_alias;
1741 }
1742 }
1743 continue;
1744 }
1745 }
1746 // A RIGHT JOIN against SQLite needs >= 3.39; warn once per
1747 // process. Postgres / no-pool builds stay silent.
1748 if emitted_right {
1749 warn_right_join_on_sqlite();
1750 }
1751 *q = outer;
1752 }
1753
1754 /// Run the SELECT and return every matching row.
1755 ///
1756 /// If `.select_related(name)` was called, a follow-up batch query
1757 /// populates `ForeignKey<U>.resolved` for each named field before
1758 /// the rows are returned.
1759 pub async fn fetch(self) -> Result<Vec<T>, sqlx::Error>
1760 where
1761 T: for<'r> sqlx::FromRow<'r, sqlx::sqlite::SqliteRow>
1762 + for<'r> sqlx::FromRow<'r, sqlx::postgres::PgRow>
1763 + HydrateRelated,
1764 {
1765 if self.only_cols.is_some() {
1766 return Err(only_with_typed_terminal_error("fetch"));
1767 }
1768 let sr_fields = self.select_related.clone();
1769 let prefetch_fields = self.prefetch_related.clone();
1770 let join_reqs = self.join_related.clone();
1771 let join_fields: Vec<String> = join_reqs.iter().map(|j| j.path.clone()).collect();
1772 // Validate join_related field names up front so a typo or an
1773 // M2M field name doesn't silently no-op (it used to render a
1774 // SELECT with no JOIN). Pre-#42 the failure mode was
1775 // `ForeignKey::resolved()` stays None and the caller debugs
1776 // the wrong thing. Now they get a typed error.
1777 validate_join_related_fields::<T>(&join_fields)?;
1778 // The turbofish on `query_as_with::<DB, _, _>` is load-bearing:
1779 // with both `sqlx-sqlite` and `sqlx-postgres` features on
1780 // sea-query-binder, `SqlxValues` implements `IntoArguments` for
1781 // both backends, so the compiler can't infer DB from the values
1782 // alone. Naming DB explicitly pins which `FromRow` bound is
1783 // checked.
1784 // Split join_fields into FK vs M2M groups up front. The
1785 // M2M branch needs parent dedup (one parent row per JOIN
1786 // combo would surface duplicate Ts to the caller); the FK
1787 // branch is one-to-one with rows.
1788 // A path whose FIRST segment is an M2M field routes to the M2M
1789 // group even when it continues with an onward FK chain
1790 // (`"tags__category"`): the junction double-join + parent dedup
1791 // live on the M2M side, and the onward FK nests into each child.
1792 let (m2m_join_fields, fk_join_fields): (Vec<String>, Vec<String>) =
1793 join_fields.iter().cloned().partition(|f| {
1794 let first = f.split("__").next().unwrap_or(f.as_str());
1795 T::M2M_RELATIONS.iter().any(|r| r.field_name == first)
1796 });
1797 let has_m2m_join = !m2m_join_fields.is_empty();
1798
1799 let mut rows = match resolve_pool::<T>(self.explicit_pool.clone(), crate::db::RouteOp::Read)
1800 {
1801 DbPool::Sqlite(pool) => {
1802 let mut q = self.build_query_for("sqlite");
1803 self.apply_join_related(&mut q);
1804 let (sql, values) = q.build_sqlx(SqliteQueryBuilder);
1805 if join_fields.is_empty() {
1806 sqlx::query_as_with::<sqlx::Sqlite, T, _>(&sql, values)
1807 .fetch_all(&pool)
1808 .await?
1809 } else if !has_m2m_join {
1810 // FK-only JOIN path: one row in → one T out.
1811 let raw_rows = sqlx::query_with::<sqlx::Sqlite, _>(&sql, values)
1812 .fetch_all(&pool)
1813 .await?;
1814 let mut typed = Vec::with_capacity(raw_rows.len());
1815 for row in &raw_rows {
1816 let mut t = <T as sqlx::FromRow<_>>::from_row(row)?;
1817 backend_sqlite::hydrate_joined_rels::<T>(&mut t, row, &fk_join_fields)?;
1818 typed.push(t);
1819 }
1820 typed
1821 } else {
1822 // Mixed (FK + M2M) or pure M2M JOIN path:
1823 // dedup parents, collect M2M children per
1824 // (parent_pk, field).
1825 let raw_rows = sqlx::query_with::<sqlx::Sqlite, _>(&sql, values)
1826 .fetch_all(&pool)
1827 .await?;
1828 dedup_decode_sqlite::<T>(&raw_rows, &fk_join_fields, &m2m_join_fields)?
1829 }
1830 }
1831 DbPool::Postgres(pool) => {
1832 let mut q = self.build_query_for("postgres");
1833 self.apply_join_related(&mut q);
1834 let (sql, values) = q.build_sqlx(PostgresQueryBuilder);
1835 if join_fields.is_empty() {
1836 sqlx::query_as_with::<sqlx::Postgres, T, _>(&sql, values)
1837 .fetch_all(&pool)
1838 .await?
1839 } else if !has_m2m_join {
1840 let raw_rows = sqlx::query_with::<sqlx::Postgres, _>(&sql, values)
1841 .fetch_all(&pool)
1842 .await?;
1843 let mut typed = Vec::with_capacity(raw_rows.len());
1844 for row in &raw_rows {
1845 let mut t = <T as sqlx::FromRow<_>>::from_row(row)?;
1846 backend_pg::hydrate_joined_rels::<T>(&mut t, row, &fk_join_fields)?;
1847 typed.push(t);
1848 }
1849 typed
1850 } else {
1851 let raw_rows = sqlx::query_with::<sqlx::Postgres, _>(&sql, values)
1852 .fetch_all(&pool)
1853 .await?;
1854 dedup_decode_pg::<T>(&raw_rows, &fk_join_fields, &m2m_join_fields)?
1855 }
1856 }
1857 };
1858 // BUG-16 step 2: wire each row's PK into its `M2M<U>` slots so
1859 // `add`/`remove`/`clear` know which parent they belong to.
1860 // No-op for models with no M2M fields.
1861 for r in &mut rows {
1862 r.set_m2m_parent_ids();
1863 }
1864 if !sr_fields.is_empty() {
1865 let pool = resolve_pool::<T>(self.explicit_pool.clone(), crate::db::RouteOp::Read);
1866 hydrate_select_related::<T>(&mut rows, &sr_fields, &pool).await?;
1867 }
1868 if !prefetch_fields.is_empty() {
1869 let pool = resolve_pool::<T>(self.explicit_pool.clone(), crate::db::RouteOp::Read);
1870 hydrate_prefetch_related::<T>(&mut rows, &prefetch_fields, &pool).await?;
1871 }
1872 Ok(rows)
1873 }
1874
1875 /// Feature 29 Phase 1 — chunked streaming via a callback.
1876 ///
1877 /// Runs the SELECT in pages of `chunk_size` rows and invokes
1878 /// `callback` once per row. Memory bound = `chunk_size *
1879 /// sizeof::<T>` instead of the full row count `fetch()` would
1880 /// buffer, so this is the right shape for million-row exports,
1881 /// migrations, and batch transforms.
1882 ///
1883 /// Deliberately NOT named `iterator()` — that name suggests a
1884 /// `Stream`-shaped return value, which would force a
1885 /// `futures-util` dep. The callback shape is idiomatic Rust,
1886 /// requires no new crates, and ships the same memory bound. A
1887 /// future `iterator()` returning `BoxStream<T>` can land later
1888 /// once `futures-util` is in the workspace for some other reason
1889 /// (likely SSE / WebSockets).
1890 ///
1891 /// Error contract: the callback may return any error type `E`.
1892 /// SQL failures become `TryForEachError::Sqlx`; callback errors
1893 /// become `TryForEachError::Callback(e)`. The first error stops
1894 /// the walk — subsequent rows are not fetched.
1895 ///
1896 /// Caveats: pages are stable only if the result set isn't being
1897 /// mutated concurrently. For consistent-snapshot iteration over
1898 /// a live table, wrap the call in a serialised-or-repeatable-read
1899 /// transaction. `select_related` and `prefetch_related` hooks
1900 /// are NOT applied on each row — `try_for_each` is intentionally
1901 /// the "raw column data, one row at a time" terminal.
1902 pub async fn try_for_each<F, E>(
1903 self,
1904 chunk_size: usize,
1905 mut callback: F,
1906 ) -> Result<(), TryForEachError<E>>
1907 where
1908 T: for<'r> sqlx::FromRow<'r, sqlx::sqlite::SqliteRow>
1909 + for<'r> sqlx::FromRow<'r, sqlx::postgres::PgRow>
1910 + HydrateRelated,
1911 F: FnMut(T) -> Result<(), E>,
1912 {
1913 let chunk_size = chunk_size.max(1);
1914 let pool = resolve_pool::<T>(self.explicit_pool.clone(), crate::db::RouteOp::Read);
1915 // gaps4 #27: honor a caller `.limit(n)` as a TOTAL cap across the whole
1916 // scan. `try_for_each` manages its own LIMIT/OFFSET paging, and it used
1917 // to overwrite the statement's LIMIT unconditionally — so a caller who
1918 // wrote `.limit(100).try_for_each(...)` silently walked the entire
1919 // table. Tracked separately because sea-query won't hand its LIMIT back.
1920 let total_cap = self.user_limit;
1921 let mut seen: u64 = 0;
1922 // Start paging from a caller-supplied `.offset(n)` (tracked via
1923 // `user_offset`) rather than 0, so rows the caller asked to skip stay
1924 // skipped — the OFFSET twin of the `user_limit` cap above.
1925 let mut offset: u64 = self.user_offset.unwrap_or(0);
1926 loop {
1927 // Never fetch more than the remaining cap allows.
1928 let this_chunk = match total_cap {
1929 Some(cap) => (chunk_size as u64).min(cap.saturating_sub(seen)),
1930 None => chunk_size as u64,
1931 };
1932 if this_chunk == 0 {
1933 break;
1934 }
1935 let mut rows: Vec<T> = match &pool {
1936 DbPool::Sqlite(pg) => {
1937 let mut q = self.build_query_for("sqlite");
1938 q.limit(this_chunk).offset(offset);
1939 let (sql, values) = q.build_sqlx(SqliteQueryBuilder);
1940 sqlx::query_as_with::<sqlx::Sqlite, T, _>(&sql, values)
1941 .fetch_all(pg)
1942 .await
1943 .map_err(TryForEachError::Sqlx)?
1944 }
1945 DbPool::Postgres(pg) => {
1946 let mut q = self.build_query_for("postgres");
1947 q.limit(this_chunk).offset(offset);
1948 let (sql, values) = q.build_sqlx(PostgresQueryBuilder);
1949 sqlx::query_as_with::<sqlx::Postgres, T, _>(&sql, values)
1950 .fetch_all(pg)
1951 .await
1952 .map_err(TryForEachError::Sqlx)?
1953 }
1954 };
1955 let fetched = rows.len();
1956 if fetched == 0 {
1957 break;
1958 }
1959 for row in rows.drain(..) {
1960 callback(row).map_err(TryForEachError::Callback)?;
1961 }
1962 seen += fetched as u64;
1963 // A short read (fewer than we asked for) means the source is
1964 // exhausted; and once we've hit the cap there's nothing more to do.
1965 if (fetched as u64) < this_chunk {
1966 break;
1967 }
1968 offset += fetched as u64;
1969 }
1970 Ok(())
1971 }
1972
1973 /// Run the SELECT with LIMIT 1 and return the first row, if any.
1974 pub async fn first(mut self) -> Result<Option<T>, sqlx::Error>
1975 where
1976 T: for<'r> sqlx::FromRow<'r, sqlx::sqlite::SqliteRow>
1977 + for<'r> sqlx::FromRow<'r, sqlx::postgres::PgRow>
1978 + HydrateRelated,
1979 {
1980 if self.only_cols.is_some() {
1981 return Err(only_with_typed_terminal_error("first"));
1982 }
1983 // Review #2: delegate to `fetch()` with LIMIT 1 so select_related,
1984 // prefetch_related, AND join_related are all hydrated. `first()`
1985 // used to build a plain query and hydrate only select_related, so
1986 // `.prefetch_related("tags").first()` returned an unprefetched row
1987 // and `.join_related("author").first()` an unresolved join — both
1988 // silently. (For a to-many `join_related`, LIMIT 1 truncates the
1989 // joined children the same way `fetch()` does with `.limit(1)`;
1990 // prefer `prefetch_related` there.)
1991 self.query.limit(1);
1992 let rows = self.fetch().await?;
1993 Ok(rows.into_iter().next())
1994 }
1995
1996 /// Return the row with the smallest value in `col_name`. Sugar
1997 /// for `order_by(col.asc()).first()`. The `earliest('created_at')`
1998 /// terminal.
1999 ///
2000 /// Takes a `&'static str` column name (same shape as
2001 /// `select_related`) so the call site stays terse:
2002 /// `.earliest("created_at")` reads naturally without spelling out
2003 /// `.asc()`.
2004 pub async fn earliest(self, col_name: &'static str) -> Result<Option<T>, sqlx::Error>
2005 where
2006 T: for<'r> sqlx::FromRow<'r, sqlx::sqlite::SqliteRow>
2007 + for<'r> sqlx::FromRow<'r, sqlx::postgres::PgRow>
2008 + HydrateRelated,
2009 {
2010 self.order_by(OrderExpr::new(col_name, false)).first().await
2011 }
2012
2013 /// Return the row with the largest value in `col_name`. Sugar
2014 /// for `order_by(col.desc()).first()`. The `latest('created_at')`
2015 /// terminal.
2016 pub async fn latest(self, col_name: &'static str) -> Result<Option<T>, sqlx::Error>
2017 where
2018 T: for<'r> sqlx::FromRow<'r, sqlx::sqlite::SqliteRow>
2019 + for<'r> sqlx::FromRow<'r, sqlx::postgres::PgRow>
2020 + HydrateRelated,
2021 {
2022 self.order_by(OrderExpr::new(col_name, true)).first().await
2023 }
2024
2025 /// Fetch many rows by their primary keys and return a
2026 /// `HashMap<T::PrimaryKey, T>` keyed by PK. The everyday companion to
2027 /// a cached list of ids — `User::objects().in_bulk(user_ids)` gives
2028 /// you direct lookup access without a second `.iter().find(...)` pass
2029 /// per id.
2030 ///
2031 /// Missing ids are silently absent from the map; callers that
2032 /// need the existence check can compare `map.len()` to
2033 /// `pks.len()`. Empty input is a no-op (returns the empty map).
2034 ///
2035 /// PK-agnostic (PK lift — was `Vec<i64>` / `HashMap<i64, T>`): the key
2036 /// is the model's `PrimaryKey` type, so i64-, String/slug-, and
2037 /// Uuid-keyed models all work. The map key requires `Hash + Eq`,
2038 /// which every standard PK type (integers, `String`, `Uuid`) satisfies.
2039 pub async fn in_bulk(
2040 self,
2041 pks: Vec<T::PrimaryKey>,
2042 ) -> Result<HashMap<T::PrimaryKey, T>, sqlx::Error>
2043 where
2044 T: for<'r> sqlx::FromRow<'r, sqlx::sqlite::SqliteRow>
2045 + for<'r> sqlx::FromRow<'r, sqlx::postgres::PgRow>
2046 + HydrateRelated,
2047 T::PrimaryKey: std::hash::Hash + Eq,
2048 {
2049 if pks.is_empty() {
2050 return Ok(HashMap::new());
2051 }
2052 let pk_name = pk_field::<T>().map(|f| f.name).unwrap_or("id");
2053 // Each PK converts to a `sea_query::Value` (the `PrimaryKey` trait
2054 // bounds `Into<sea_query::Value>`); wrap as a `SimpleExpr` for the
2055 // IN-list so any PK shape binds correctly.
2056 let pk_pred: Predicate<T> = Predicate::new(
2057 Expr::col(Alias::new(pk_name)).is_in(
2058 pks.into_iter()
2059 .map(|p| sea_query::SimpleExpr::Value(p.into())),
2060 ),
2061 );
2062 let rows = self.filter(pk_pred).fetch().await?;
2063 let mut out: HashMap<T::PrimaryKey, T> = HashMap::with_capacity(rows.len());
2064 for row in rows {
2065 out.insert(row.primary_key(), row);
2066 }
2067 Ok(out)
2068 }
2069
2070 /// Return the database's execution plan for this query as a
2071 /// plain-text string. Doesn't run the underlying query — just
2072 /// asks the DB how it would be executed.
2073 ///
2074 /// Backend dispatch:
2075 ///
2076 /// - SQLite: `EXPLAIN QUERY PLAN <sql>` — returns the planner's
2077 /// nested loop hierarchy, one row per access step.
2078 /// - Postgres: `EXPLAIN <sql>` — returns the default text plan.
2079 /// For machine-readable output use raw sqlx with
2080 /// `EXPLAIN (FORMAT JSON)`; the framework defaults to text
2081 /// because most callers want eyeball-able output.
2082 ///
2083 /// Lines are joined with newlines. The returned string is what a
2084 /// developer would paste into a debugger or a perf-review issue.
2085 pub async fn explain(self) -> Result<String, sqlx::Error> {
2086 // Annotations are part of the built query (they render inside
2087 // build_query_for), so the plan below includes them — but a
2088 // poisoned annotation (unknown relation) must fail loudly
2089 // here, not silently vanish from the plan.
2090 self.check_annotations()?;
2091 let pool = resolve_pool::<T>(self.explicit_pool.clone(), crate::db::RouteOp::Read);
2092 let backend = pool.backend_name();
2093 let q = self.build_query_for(backend);
2094 match pool {
2095 DbPool::Sqlite(pool) => {
2096 let (sql, vals) = q.build_sqlx(SqliteQueryBuilder);
2097 let explain_sql = format!("EXPLAIN QUERY PLAN {sql}");
2098 let rows = sqlx::query_with::<sqlx::Sqlite, _>(&explain_sql, vals)
2099 .fetch_all(&pool)
2100 .await?;
2101 let mut out = String::new();
2102 for row in &rows {
2103 use sqlx::Row;
2104 // SQLite returns: id, parent, notused, detail.
2105 // The `detail` column is the human-readable step.
2106 let detail: String = row.try_get("detail")?;
2107 if !out.is_empty() {
2108 out.push('\n');
2109 }
2110 out.push_str(&detail);
2111 }
2112 Ok(out)
2113 }
2114 DbPool::Postgres(pool) => {
2115 let (sql, vals) = q.build_sqlx(PostgresQueryBuilder);
2116 let explain_sql = format!("EXPLAIN {sql}");
2117 let rows = sqlx::query_with::<sqlx::Postgres, _>(&explain_sql, vals)
2118 .fetch_all(&pool)
2119 .await?;
2120 let mut out = String::new();
2121 for row in &rows {
2122 use sqlx::Row;
2123 // Postgres EXPLAIN returns one column named
2124 // "QUERY PLAN", one row per line of the plan.
2125 let line: String = row.try_get("QUERY PLAN")?;
2126 if !out.is_empty() {
2127 out.push('\n');
2128 }
2129 out.push_str(&line);
2130 }
2131 Ok(out)
2132 }
2133 }
2134 }
2135
2136 /// Run `SELECT COUNT(*)` against the same FROM + WHERE.
2137 ///
2138 /// Reshapes the query rather than wrapping the existing SELECT: the
2139 /// projection becomes `COUNT(*)` and LIMIT/OFFSET drop away. ORDER
2140 /// BY is harmless on a scalar aggregate and is left in place. The
2141 /// row type is `(i64,)` so the FromRow constraint comes from sqlx's
2142 /// tuple impl rather than the user struct — count() doesn't need
2143 /// T's FromRow bounds.
2144 pub async fn count(self) -> Result<i64, sqlx::Error> {
2145 let pool = resolve_pool::<T>(self.explicit_pool.clone(), crate::db::RouteOp::Read);
2146 let backend = pool.backend_name();
2147 // Build the dialect-appropriate filtered query first, then
2148 // rebuild as COUNT. Doing it in this order keeps the predicate
2149 // walk pluggable per backend without duplicating the COUNT
2150 // rewrite logic across branches.
2151 let mut rebuilt = self.build_query_for(backend);
2152 rebuilt.clear_selects();
2153 // Postgres rejects `"*"` as a quoted identifier (SQLite tolerates
2154 // it); use sea_query's Asterisk token which renders bare `*`
2155 // on both backends.
2156 rebuilt.expr(Func::count(Expr::col(sea_query::Asterisk)));
2157 rebuilt.reset_limit();
2158 rebuilt.reset_offset();
2159
2160 match pool {
2161 DbPool::Sqlite(pool) => {
2162 let (sql, values) = rebuilt.build_sqlx(SqliteQueryBuilder);
2163 let (n,): (i64,) = sqlx::query_as_with::<sqlx::Sqlite, (i64,), _>(&sql, values)
2164 .fetch_one(&pool)
2165 .await?;
2166 Ok(n)
2167 }
2168 DbPool::Postgres(pool) => {
2169 let (sql, values) = rebuilt.build_sqlx(PostgresQueryBuilder);
2170 let (n,): (i64,) = sqlx::query_as_with::<sqlx::Postgres, (i64,), _>(&sql, values)
2171 .fetch_one(&pool)
2172 .await?;
2173 Ok(n)
2174 }
2175 }
2176 }
2177
2178 /// Return whether any row matches.
2179 ///
2180 /// M1 keeps the simple form: add LIMIT 1, fetch, check non-empty.
2181 /// A later milestone may swap the projection for `SELECT 1` to
2182 /// skip column materialisation.
2183 pub async fn exists(self) -> Result<bool, sqlx::Error>
2184 where
2185 T: for<'r> sqlx::FromRow<'r, sqlx::sqlite::SqliteRow>
2186 + for<'r> sqlx::FromRow<'r, sqlx::postgres::PgRow>
2187 + HydrateRelated,
2188 {
2189 let rows = self.limit(1).fetch().await?;
2190 Ok(!rows.is_empty())
2191 }
2192
2193 /// `.get()` — the exactly-one terminal.
2194 ///
2195 /// Returns `Ok(row)` when the filter chain matches exactly one
2196 /// row. The two not-exactly-one cases each get their own
2197 /// `GetError` variant so the caller can branch deliberately:
2198 ///
2199 /// - [`GetError::NotFound`] — zero rows matched. The right
2200 /// choice for "fetch the row this user just clicked on; 404
2201 /// if it's gone."
2202 /// - [`GetError::MultipleObjectsReturned`] — more than one row
2203 /// matched. The right choice for filters that should be
2204 /// uniquely-keyed (e.g. `.filter(user::EMAIL.eq("..."))`
2205 /// when email has a UNIQUE constraint); a result of 2+ is a
2206 /// data-integrity bug worth crashing on.
2207 /// - The underlying sqlx error wraps as [`GetError::Sqlx`].
2208 ///
2209 /// Internally this issues `SELECT ... LIMIT 2` — the cheapest
2210 /// way to distinguish "one row" from "many." The second row, if
2211 /// it exists, isn't materialised beyond the bare FromRow call.
2212 ///
2213 /// ```ignore
2214 /// match Post::objects().filter(post::ID.eq(42)).get().await {
2215 /// Ok(p) => /* render */,
2216 /// Err(GetError::NotFound) => /* 404 */,
2217 /// Err(GetError::MultipleObjectsReturned) => unreachable!("ID is unique"),
2218 /// Err(GetError::Sqlx(e)) => /* 500 */,
2219 /// }
2220 /// ```
2221 pub async fn get(self) -> Result<T, GetError>
2222 where
2223 T: for<'r> sqlx::FromRow<'r, sqlx::sqlite::SqliteRow>
2224 + for<'r> sqlx::FromRow<'r, sqlx::postgres::PgRow>
2225 + HydrateRelated,
2226 {
2227 let mut rows = self.limit(2).fetch().await.map_err(GetError::Sqlx)?;
2228 match rows.len() {
2229 0 => Err(GetError::NotFound),
2230 1 => Ok(rows.pop().unwrap()),
2231 _ => Err(GetError::MultipleObjectsReturned),
2232 }
2233 }
2234
2235 // =====================================================================
2236 // Postgres-only terminals (Phase 4.1).
2237 //
2238 // Models with Postgres-only field types (`Vec<T>` arrays, the future
2239 // Hstore / CIDR / FullTextSearch types) can't satisfy the dual
2240 // FromRow bound on `fetch` / `first` / `count` / `exists`. These
2241 // `_pg` variants bound on `FromRow<PgRow>` alone, take the pool as
2242 // an argument, and skip the dispatch — the call site explicitly
2243 // says "this model is Postgres-only."
2244 //
2245 // For models with portable fields, the existing `fetch` etc. stay
2246 // the recommended call: they pick up the ambient pool and route
2247 // through `.on(&pool)` / `.on_pg(&pool)` overrides exactly as
2248 // Phase 2.5 documented.
2249 // =====================================================================
2250
2251 // =====================================================================
2252 // Write terminals — DELETE and UPDATE.
2253 //
2254 // Both apply the accumulated filter predicates as the WHERE clause,
2255 // dispatch to the resolved pool's backend, and return the affected-
2256 // rows count from sqlx. No row materialisation — DELETE is keyless,
2257 // and UPDATE doesn't do a RETURNING read-back at v1 (use
2258 // `.filter(...).fetch()` after a write if you need the updated
2259 // rows back).
2260 //
2261 // **Without a `.filter(...)`, both terminals affect every row in
2262 // the table.** That mirrors raw SQL semantics; the type system
2263 // can't distinguish "I forgot the filter" from "I really meant to
2264 // truncate." Users protecting against accidental full-table writes
2265 // wrap their callers or assert a row count via `.count()` first.
2266 // =====================================================================
2267
2268 /// Project the query to only the named columns, returning a
2269 /// vector of `serde_json::Value::Object` rows instead of typed
2270 /// `T` instances. The columns-projection terminal: `values('id', 'title')`.
2271 ///
2272 /// Use when a list view only needs a few fields — skipping the
2273 /// 50KB body BLOB on every Post saves both memory and the
2274 /// FromRow hydration overhead. Each returned `Value` is an
2275 /// object keyed by the requested column names, with values
2276 /// typed per the column's declared SqlType (integers stay
2277 /// integers, booleans stay booleans, dates render as ISO
2278 /// strings).
2279 ///
2280 /// Unknown column names fail loudly with
2281 /// `sqlx::Error::Protocol` naming the offending column.
2282 /// Composes with `filter`, `exclude`, `order_by`, `limit`,
2283 /// `offset` exactly the way the typed terminals do.
2284 ///
2285 /// ```rust,ignore
2286 /// let rows = Post::objects()
2287 /// .filter(post::PUBLISHED.eq(true))
2288 /// .order_by(post::ID.desc())
2289 /// .values(&["id", "title"])
2290 /// .await?;
2291 /// // [ { "id": 3, "title": "c" }, ... ]
2292 /// ```
2293 pub async fn values(self, columns: &[&str]) -> Result<Vec<JsonValue>, sqlx::Error> {
2294 // Gap #46 follow-up: if any name uses `__` traversal
2295 // (`author__id`), route to the JOIN-aware path that builds
2296 // nested per-relation JSON objects. The unbranched path
2297 // below stays byte-for-byte identical for the common
2298 // parent-cols-only case.
2299 if columns.iter().any(|c| c.contains("__")) {
2300 return self.values_with_traversal(columns).await;
2301 }
2302 let meta = crate::migrate::ModelMeta::for_::<T>();
2303 // Resolve every requested name against the model's metadata
2304 // up front so an unknown column errors before any SQL runs.
2305 let mut chosen: Vec<&crate::migrate::Column> = Vec::with_capacity(columns.len());
2306 for name in columns {
2307 let col = meta
2308 .fields
2309 .iter()
2310 .find(|c| c.name == *name)
2311 .ok_or_else(|| {
2312 sqlx::Error::Protocol(format!(
2313 "umbral::orm::values: unknown column `{}` on model `{}`",
2314 name,
2315 T::NAME
2316 ))
2317 })?;
2318 chosen.push(col);
2319 }
2320 let pool = resolve_pool::<T>(self.explicit_pool.clone(), crate::db::RouteOp::Read);
2321 let backend = pool.backend_name();
2322 // Build the base query (predicates + ORDER BY) then swap its
2323 // SELECT list for only the requested columns.
2324 let mut q = self.build_query_for(backend);
2325 q.clear_selects();
2326 for col in &chosen {
2327 q.column(Alias::new(col.name.as_str()));
2328 }
2329 match pool {
2330 DbPool::Sqlite(pool) => {
2331 let (sql, vals) = q.build_sqlx(SqliteQueryBuilder);
2332 let rows = sqlx::query_with::<sqlx::Sqlite, _>(&sql, vals)
2333 .fetch_all(&pool)
2334 .await?;
2335 let mut out: Vec<JsonValue> = Vec::with_capacity(rows.len());
2336 for row in &rows {
2337 let mut obj = serde_json::Map::with_capacity(chosen.len());
2338 for col in &chosen {
2339 let v = crate::orm::dynamic::decode_to_json(row, col)?;
2340 obj.insert(col.name.clone(), v);
2341 }
2342 out.push(JsonValue::Object(obj));
2343 }
2344 Ok(out)
2345 }
2346 DbPool::Postgres(pool) => {
2347 let (sql, vals) = q.build_sqlx(PostgresQueryBuilder);
2348 let rows = sqlx::query_with::<sqlx::Postgres, _>(&sql, vals)
2349 .fetch_all(&pool)
2350 .await?;
2351 let mut out: Vec<JsonValue> = Vec::with_capacity(rows.len());
2352 for row in &rows {
2353 let mut obj = serde_json::Map::with_capacity(chosen.len());
2354 for col in &chosen {
2355 let v = crate::orm::dynamic::decode_pg_to_json(row, col)?;
2356 obj.insert(col.name.clone(), v);
2357 }
2358 out.push(JsonValue::Object(obj));
2359 }
2360 Ok(out)
2361 }
2362 }
2363 }
2364
2365 /// `.values("author__name")`-style traversal. One-hop only at
2366 /// v1 (`a__b`, not `a__b__c` — that fails loudly so the user
2367 /// doesn't get a silent partial result). Emits one LEFT JOIN
2368 /// per distinct relation referenced, aliases child columns as
2369 /// `<rel>__<col>`, and returns each row as a nested JSON
2370 /// object: `{id, title, author: {id, name}, editor: {id}}`.
2371 ///
2372 /// A LEFT JOIN miss (nullable FK pointing at nothing) maps the
2373 /// whole relation key to `Value::Null` rather than a nested
2374 /// object full of nulls — caller code that branches on
2375 /// `obj["author"].is_null()` works naturally.
2376 ///
2377 /// Validates every name up front so a typo errors before any
2378 /// SQL runs. Unknown parent col / non-FK relation name /
2379 /// unknown child col / deeper-than-one-hop path all surface
2380 /// distinct messages.
2381 async fn values_with_traversal(self, columns: &[&str]) -> Result<Vec<JsonValue>, sqlx::Error> {
2382 use sea_query::{Expr, Query};
2383 let meta = crate::migrate::ModelMeta::for_::<T>();
2384 let registered = crate::migrate::registered_models();
2385
2386 // Split each column name into (relation, child_col) or
2387 // (None, parent_col). Reject paths with more than one `__`
2388 // hop — nested traversal across two relation layers is a
2389 // separate piece of work (it'd need to chain JOINs and
2390 // build doubly-nested JSON).
2391 let mut parent_cols: Vec<String> = Vec::new();
2392 let mut per_rel: std::collections::BTreeMap<String, Vec<String>> =
2393 std::collections::BTreeMap::new();
2394 for raw in columns {
2395 let mut parts = raw.splitn(3, "__");
2396 let first = parts.next().unwrap_or("");
2397 let second = parts.next();
2398 let third = parts.next();
2399 if third.is_some() {
2400 return Err(sqlx::Error::Protocol(format!(
2401 "umbral::orm::values: nested `{raw}` is not supported in v1 \
2402 (one-hop only — `a__b`, not `a__b__c`)"
2403 )));
2404 }
2405 match second {
2406 Some(child) => {
2407 per_rel
2408 .entry(first.to_string())
2409 .or_default()
2410 .push(child.to_string());
2411 }
2412 None => parent_cols.push(first.to_string()),
2413 }
2414 }
2415
2416 // Validate every parent name against T::FIELDS.
2417 for name in &parent_cols {
2418 if !meta.fields.iter().any(|c| c.name == *name) {
2419 return Err(sqlx::Error::Protocol(format!(
2420 "umbral::orm::values: unknown column `{name}` on model `{}`",
2421 T::NAME
2422 )));
2423 }
2424 }
2425
2426 // Validate every relation + child trio. Build a struct per
2427 // relation that the SQL/decoder loops below need.
2428 struct RelInfo<'a> {
2429 rel_name: String,
2430 related_table: &'a str,
2431 related_pk: &'a crate::migrate::Column,
2432 child_cols: Vec<&'a crate::migrate::Column>,
2433 }
2434 let mut rel_infos: Vec<RelInfo<'_>> = Vec::with_capacity(per_rel.len());
2435 for (rel_name, child_names) in &per_rel {
2436 let fk_field = meta.fields.iter().find(|c| c.name == *rel_name);
2437 let Some(fk_field) = fk_field else {
2438 return Err(sqlx::Error::Protocol(format!(
2439 "umbral::orm::values: unknown relation `{rel_name}` on model `{}` \
2440 (used in `{rel_name}__...`)",
2441 T::NAME
2442 )));
2443 };
2444 let Some(related_table) = fk_field.fk_target.as_deref() else {
2445 return Err(sqlx::Error::Protocol(format!(
2446 "umbral::orm::values: field `{rel_name}` on `{}` is not a foreign \
2447 key — `__` traversal only works through FK fields",
2448 T::NAME
2449 )));
2450 };
2451 let Some(related_meta) = registered.iter().find(|m| m.table == related_table) else {
2452 return Err(sqlx::Error::Protocol(format!(
2453 "umbral::orm::values: related model for table `{related_table}` \
2454 is not registered"
2455 )));
2456 };
2457 let Some(related_pk) = related_meta.fields.iter().find(|c| c.primary_key) else {
2458 return Err(sqlx::Error::Protocol(format!(
2459 "umbral::orm::values: related model `{related_table}` has no \
2460 primary key column"
2461 )));
2462 };
2463 let mut child_cols: Vec<&crate::migrate::Column> =
2464 Vec::with_capacity(child_names.len());
2465 for name in child_names {
2466 let col = related_meta
2467 .fields
2468 .iter()
2469 .find(|c| c.name == *name)
2470 .ok_or_else(|| {
2471 sqlx::Error::Protocol(format!(
2472 "umbral::orm::values: unknown child column `{name}` on \
2473 related model `{related_table}` (full path `{rel_name}__{name}`)"
2474 ))
2475 })?;
2476 child_cols.push(col);
2477 }
2478 rel_infos.push(RelInfo {
2479 rel_name: rel_name.clone(),
2480 related_table,
2481 related_pk,
2482 child_cols,
2483 });
2484 }
2485
2486 // Build the SQL. Subquery-wrap the parent (WHERE / ORDER
2487 // BY / LIMIT all stay scoped to it) so JOIN'd tables can't
2488 // shadow bare-column predicates — same trick
2489 // apply_join_related uses.
2490 let pool = resolve_pool::<T>(self.explicit_pool.clone(), crate::db::RouteOp::Read);
2491 let backend = pool.backend_name();
2492 let inner = self.build_query_for(backend);
2493 let parent_alias = Alias::new("__p");
2494 let mut outer = Query::select();
2495 outer.from_subquery(inner, parent_alias.clone());
2496 // Outer SELECT: parent cols (bare), aliased child cols.
2497 for name in &parent_cols {
2498 outer.expr_as(
2499 Expr::col((parent_alias.clone(), Alias::new(name.as_str()))),
2500 Alias::new(name.as_str()),
2501 );
2502 }
2503 for info in &rel_infos {
2504 let join_alias = Alias::new(format!("__j_{}", info.rel_name));
2505 outer.join_as(
2506 sea_query::JoinType::LeftJoin,
2507 crate::db::router::schema_qualified_table(info.related_table),
2508 join_alias.clone(),
2509 Expr::col((parent_alias.clone(), Alias::new(info.rel_name.as_str()))).equals((
2510 join_alias.clone(),
2511 Alias::new(info.related_pk.name.as_str()),
2512 )),
2513 );
2514 // Always include the related PK alias so the decoder
2515 // can detect a LEFT JOIN miss → emit Value::Null for
2516 // the whole relation. Plus every requested child col.
2517 let pk_alias = format!("{}__{}", info.rel_name, info.related_pk.name);
2518 outer.expr_as(
2519 Expr::col((
2520 join_alias.clone(),
2521 Alias::new(info.related_pk.name.as_str()),
2522 )),
2523 Alias::new(pk_alias),
2524 );
2525 for col in &info.child_cols {
2526 let alias = format!("{}__{}", info.rel_name, col.name);
2527 outer.expr_as(
2528 Expr::col((join_alias.clone(), Alias::new(col.name.as_str()))),
2529 Alias::new(alias),
2530 );
2531 }
2532 }
2533
2534 // Execute + decode. Per-backend dispatch.
2535 match pool {
2536 DbPool::Sqlite(pool) => {
2537 let (sql, vals) = outer.build_sqlx(SqliteQueryBuilder);
2538 let rows = sqlx::query_with::<sqlx::Sqlite, _>(&sql, vals)
2539 .fetch_all(&pool)
2540 .await?;
2541 let mut out: Vec<JsonValue> = Vec::with_capacity(rows.len());
2542 for row in &rows {
2543 let mut obj = serde_json::Map::new();
2544 // Parent cols decode by their bare alias name.
2545 for name in &parent_cols {
2546 if let Some(col) = meta.fields.iter().find(|c| c.name == *name) {
2547 let v = crate::orm::dynamic::decode_to_json_aliased(row, col, name)?;
2548 obj.insert(name.clone(), v);
2549 }
2550 }
2551 // Per-relation nested object — `null` on LEFT JOIN miss.
2552 for info in &rel_infos {
2553 let pk_alias = format!("{}__{}", info.rel_name, info.related_pk.name);
2554 let pk_is_null =
2555 backend_sqlite::joined_pk_is_null(row, info.related_pk, &pk_alias);
2556 if pk_is_null {
2557 obj.insert(info.rel_name.clone(), JsonValue::Null);
2558 continue;
2559 }
2560 let mut nested = serde_json::Map::with_capacity(info.child_cols.len());
2561 for col in &info.child_cols {
2562 let alias = format!("{}__{}", info.rel_name, col.name);
2563 let v = crate::orm::dynamic::decode_to_json_aliased(row, col, &alias)?;
2564 nested.insert(col.name.clone(), v);
2565 }
2566 obj.insert(info.rel_name.clone(), JsonValue::Object(nested));
2567 }
2568 out.push(JsonValue::Object(obj));
2569 }
2570 Ok(out)
2571 }
2572 DbPool::Postgres(pool) => {
2573 let (sql, vals) = outer.build_sqlx(PostgresQueryBuilder);
2574 let rows = sqlx::query_with::<sqlx::Postgres, _>(&sql, vals)
2575 .fetch_all(&pool)
2576 .await?;
2577 let mut out: Vec<JsonValue> = Vec::with_capacity(rows.len());
2578 for row in &rows {
2579 let mut obj = serde_json::Map::new();
2580 for name in &parent_cols {
2581 if let Some(col) = meta.fields.iter().find(|c| c.name == *name) {
2582 let v = crate::orm::dynamic::decode_pg_to_json_aliased(row, col, name)?;
2583 obj.insert(name.clone(), v);
2584 }
2585 }
2586 for info in &rel_infos {
2587 let pk_alias = format!("{}__{}", info.rel_name, info.related_pk.name);
2588 let pk_is_null =
2589 backend_pg::joined_pk_is_null(row, info.related_pk, &pk_alias);
2590 if pk_is_null {
2591 obj.insert(info.rel_name.clone(), JsonValue::Null);
2592 continue;
2593 }
2594 let mut nested = serde_json::Map::with_capacity(info.child_cols.len());
2595 for col in &info.child_cols {
2596 let alias = format!("{}__{}", info.rel_name, col.name);
2597 let v =
2598 crate::orm::dynamic::decode_pg_to_json_aliased(row, col, &alias)?;
2599 nested.insert(col.name.clone(), v);
2600 }
2601 obj.insert(info.rel_name.clone(), JsonValue::Object(nested));
2602 }
2603 out.push(JsonValue::Object(obj));
2604 }
2605 Ok(out)
2606 }
2607 }
2608 }
2609
2610 /// Single-row aggregate. Runs `SELECT AGG(col) AS name, ...` with
2611 /// the QuerySet's accumulated WHERE clause (ORDER BY / LIMIT /
2612 /// OFFSET are dropped — they make no sense over an aggregate
2613 /// without GROUP BY).
2614 ///
2615 /// Returns a `serde_json::Value::Object` keyed by the supplied
2616 /// names. COUNT comes back as an integer; AVG as a float; SUM /
2617 /// MAX / MIN inherit the source column's declared type.
2618 ///
2619 /// ```rust,ignore
2620 /// use umbral::orm::Aggregate;
2621 /// let summary = Post::objects()
2622 /// .filter(post::PUBLISHED.eq(true))
2623 /// .aggregate(&[
2624 /// ("count", Aggregate::count()),
2625 /// ("total", Aggregate::sum("view_count")),
2626 /// ])
2627 /// .await?;
2628 /// // { "count": 42, "total": 9999 }
2629 /// ```
2630 pub async fn aggregate(
2631 self,
2632 aggs: &[(&str, crate::orm::Aggregate)],
2633 ) -> Result<JsonValue, sqlx::Error> {
2634 let meta = crate::migrate::ModelMeta::for_::<T>();
2635 // Validate every aggregate's source column exists.
2636 for (name, agg) in aggs {
2637 if let Some(col) = agg.source_column()
2638 && !meta.fields.iter().any(|c| c.name == col)
2639 {
2640 return Err(sqlx::Error::Protocol(format!(
2641 "umbral::orm::aggregate: unknown column `{}` on model `{}` for aggregate `{}`",
2642 col,
2643 T::NAME,
2644 name
2645 )));
2646 }
2647 }
2648 let pool = resolve_pool::<T>(self.explicit_pool.clone(), crate::db::RouteOp::Read);
2649 let backend = pool.backend_name();
2650 let mut q = self.build_query_for(backend);
2651 q.clear_selects();
2652 q.reset_limit();
2653 q.reset_offset();
2654 for (name, agg) in aggs {
2655 q.expr_as(agg.to_simple_expr(), Alias::new(*name));
2656 }
2657 match pool {
2658 DbPool::Sqlite(pool) => {
2659 let (sql, vals) = q.build_sqlx(SqliteQueryBuilder);
2660 let row = sqlx::query_with::<sqlx::Sqlite, _>(&sql, vals)
2661 .fetch_one(&pool)
2662 .await?;
2663 let mut obj = serde_json::Map::with_capacity(aggs.len());
2664 for (name, agg) in aggs {
2665 let source_ty = agg
2666 .source_column()
2667 .and_then(|c| meta.fields.iter().find(|f| f.name == c).map(|f| f.ty));
2668 obj.insert(
2669 name.to_string(),
2670 backend_sqlite::decode_agg(&row, name, agg, source_ty)?,
2671 );
2672 }
2673 Ok(JsonValue::Object(obj))
2674 }
2675 DbPool::Postgres(pool) => {
2676 let (sql, vals) = q.build_sqlx(PostgresQueryBuilder);
2677 let row = sqlx::query_with::<sqlx::Postgres, _>(&sql, vals)
2678 .fetch_one(&pool)
2679 .await?;
2680 let mut obj = serde_json::Map::with_capacity(aggs.len());
2681 for (name, agg) in aggs {
2682 let source_ty = agg
2683 .source_column()
2684 .and_then(|c| meta.fields.iter().find(|f| f.name == c).map(|f| f.ty));
2685 obj.insert(
2686 name.to_string(),
2687 backend_pg::decode_agg(&row, name, agg, source_ty)?,
2688 );
2689 }
2690 Ok(JsonValue::Object(obj))
2691 }
2692 }
2693 }
2694
2695 /// Grouped aggregate. Runs `SELECT <group_cols>, AGG(col) AS name,
2696 /// ... GROUP BY <group_cols>` with the accumulated WHERE clause.
2697 ///
2698 /// Returns one `Value::Object` per group, with both the group
2699 /// columns and each named aggregate as fields. Group columns are
2700 /// decoded per their declared SqlType (so an integer
2701 /// `author_id` stays a JSON number).
2702 ///
2703 /// ```rust,ignore
2704 /// let by_author = Post::objects()
2705 /// .annotate(&["author_id"], &[("count", Aggregate::count())])
2706 /// .await?;
2707 /// // [ { "author_id": 1, "count": 3 }, { "author_id": 2, "count": 2 } ]
2708 /// ```
2709 /// [`Self::annotate`], deserialized into a struct of your own (gaps3 #29).
2710 ///
2711 /// `annotate` returns `Vec<serde_json::Value>` — a GROUP BY rollup you then
2712 /// hand-decode. This gives you the rows typed:
2713 ///
2714 /// ```ignore
2715 /// #[derive(Deserialize)]
2716 /// struct ByAuthor { author_id: i64, posts: i64 }
2717 ///
2718 /// let rows: Vec<ByAuthor> = Post::objects()
2719 /// .annotate_as::<ByAuthor>(&["author_id"], &[("posts", Aggregate::count())])
2720 /// .await?;
2721 /// ```
2722 pub async fn annotate_as<R>(
2723 self,
2724 group_cols: &[&str],
2725 aggs: &[(&str, crate::orm::Aggregate)],
2726 ) -> Result<Vec<R>, sqlx::Error>
2727 where
2728 R: serde::de::DeserializeOwned,
2729 {
2730 let rows = self.annotate(group_cols, aggs).await?;
2731 rows.into_iter()
2732 .map(|v| {
2733 serde_json::from_value::<R>(v)
2734 .map_err(|e| sqlx::Error::Protocol(format!("annotate_as: {e}")))
2735 })
2736 .collect()
2737 }
2738
2739 pub async fn annotate(
2740 self,
2741 group_cols: &[&str],
2742 aggs: &[(&str, crate::orm::Aggregate)],
2743 ) -> Result<Vec<JsonValue>, sqlx::Error> {
2744 let meta = crate::migrate::ModelMeta::for_::<T>();
2745 // Resolve group columns up front so unknown names fail
2746 // before any SQL runs.
2747 let mut chosen_groups: Vec<&crate::migrate::Column> = Vec::with_capacity(group_cols.len());
2748 for name in group_cols {
2749 let col = meta
2750 .fields
2751 .iter()
2752 .find(|c| c.name == *name)
2753 .ok_or_else(|| {
2754 sqlx::Error::Protocol(format!(
2755 "umbral::orm::annotate: unknown group column `{}` on model `{}`",
2756 name,
2757 T::NAME
2758 ))
2759 })?;
2760 chosen_groups.push(col);
2761 }
2762 // Validate aggregate source columns.
2763 for (name, agg) in aggs {
2764 if let Some(col) = agg.source_column()
2765 && !meta.fields.iter().any(|c| c.name == col)
2766 {
2767 return Err(sqlx::Error::Protocol(format!(
2768 "umbral::orm::annotate: unknown column `{}` on model `{}` for aggregate `{}`",
2769 col,
2770 T::NAME,
2771 name
2772 )));
2773 }
2774 }
2775 let pool = resolve_pool::<T>(self.explicit_pool.clone(), crate::db::RouteOp::Read);
2776 let backend = pool.backend_name();
2777 let mut q = self.build_query_for(backend);
2778 q.clear_selects();
2779 // GROUP BY columns appear in the SELECT list AND the GROUP BY
2780 // clause. Aggregates only in the SELECT.
2781 for col in &chosen_groups {
2782 q.column(Alias::new(col.name.as_str()));
2783 q.add_group_by([sea_query::SimpleExpr::Column(sea_query::ColumnRef::Column(
2784 Alias::new(col.name.as_str()).into_iden(),
2785 ))]);
2786 }
2787 for (name, agg) in aggs {
2788 q.expr_as(agg.to_simple_expr(), Alias::new(*name));
2789 }
2790 match pool {
2791 DbPool::Sqlite(pool) => {
2792 let (sql, vals) = q.build_sqlx(SqliteQueryBuilder);
2793 let rows = sqlx::query_with::<sqlx::Sqlite, _>(&sql, vals)
2794 .fetch_all(&pool)
2795 .await?;
2796 let mut out: Vec<JsonValue> = Vec::with_capacity(rows.len());
2797 for row in &rows {
2798 let mut obj = serde_json::Map::with_capacity(chosen_groups.len() + aggs.len());
2799 for col in &chosen_groups {
2800 obj.insert(
2801 col.name.clone(),
2802 crate::orm::dynamic::decode_to_json(row, col)?,
2803 );
2804 }
2805 for (name, agg) in aggs {
2806 let source_ty = agg
2807 .source_column()
2808 .and_then(|c| meta.fields.iter().find(|f| f.name == c).map(|f| f.ty));
2809 obj.insert(
2810 name.to_string(),
2811 backend_sqlite::decode_agg(row, name, agg, source_ty)?,
2812 );
2813 }
2814 out.push(JsonValue::Object(obj));
2815 }
2816 Ok(out)
2817 }
2818 DbPool::Postgres(pool) => {
2819 let (sql, vals) = q.build_sqlx(PostgresQueryBuilder);
2820 let rows = sqlx::query_with::<sqlx::Postgres, _>(&sql, vals)
2821 .fetch_all(&pool)
2822 .await?;
2823 let mut out: Vec<JsonValue> = Vec::with_capacity(rows.len());
2824 for row in &rows {
2825 let mut obj = serde_json::Map::with_capacity(chosen_groups.len() + aggs.len());
2826 for col in &chosen_groups {
2827 obj.insert(
2828 col.name.clone(),
2829 crate::orm::dynamic::decode_pg_to_json(row, col)?,
2830 );
2831 }
2832 for (name, agg) in aggs {
2833 let source_ty = agg
2834 .source_column()
2835 .and_then(|c| meta.fields.iter().find(|f| f.name == c).map(|f| f.ty));
2836 obj.insert(
2837 name.to_string(),
2838 backend_pg::decode_agg(row, name, agg, source_ty)?,
2839 );
2840 }
2841 out.push(JsonValue::Object(obj));
2842 }
2843 Ok(out)
2844 }
2845 }
2846 }
2847
2848 /// The chainable `annotate(alias=Agg("relation"))`: attach a
2849 /// related-aggregate annotation to this QuerySet. The annotation
2850 /// is **query-builder state** — it renders as a correlated scalar
2851 /// subquery inside the one SELECT every terminal builds, so it
2852 /// composes with `.filter` / `.order_by` / `.limit`, stacks with
2853 /// further annotations, and shows up in [`Self::explain`] /
2854 /// [`Self::to_sql`] out of the box. Never a side query, never an
2855 /// N+1.
2856 ///
2857 /// `relation` names a `ReverseSet` relation on the model
2858 /// (`#[umbral(reverse_fk = "...")]`), the same names
2859 /// `prefetch_related` accepts. When no declared relation matches,
2860 /// the resolver AUTO-DISCOVERS the relation (gaps2 #45): it scans
2861 /// the model registry for any child whose FK targets this parent's
2862 /// table and matches `relation` against the child's conventional
2863 /// name forms (table name, `snake_case` / lowercase struct name,
2864 /// any of those with a `_set` suffix). Declared relations always
2865 /// take precedence; an ambiguous auto-match (two children, or a
2866 /// child with two FKs to this parent) poisons the annotation with
2867 /// an error that names the candidates and points at the
2868 /// `#[umbral(reverse_fk = "...")]` escape hatch. Any
2869 /// [`crate::orm::Aggregate`] works; non-count aggregates name a
2870 /// column on the CHILD model:
2871 ///
2872 /// ```rust,ignore
2873 /// let rows = Plugin::objects()
2874 /// .filter(plugin::MODERATION.eq("approved"))
2875 /// .annotate_count("comment_set") // COUNT(*)
2876 /// .annotate_related("rating_avg", "review_set", Aggregate::avg("rating"))
2877 /// .fetch_annotated()
2878 /// .await?; // Vec<(Plugin, Map<alias, value>)>
2879 /// ```
2880 ///
2881 /// An unknown relation name doesn't panic the (infallible)
2882 /// builder — it poisons the annotation, and every fallible
2883 /// consumer (`fetch_annotated`, `explain`) reports it loudly.
2884 /// v1 caveats: child rows aggregate unconditionally — a
2885 /// child-side predicate (a filtered count)
2886 /// and child soft-delete awareness are tracked follow-ups
2887 /// (gaps2 #39).
2888 pub fn annotate_related(
2889 mut self,
2890 alias: &str,
2891 relation: &str,
2892 agg: crate::orm::Aggregate,
2893 ) -> Self {
2894 let rev_spec = T::REVERSE_FK_RELATIONS
2895 .iter()
2896 .find(|r| r.field_name == relation);
2897 let m2m_spec = T::M2M_RELATIONS.iter().find(|r| r.field_name == relation);
2898
2899 let pk = T::FIELDS
2900 .iter()
2901 .find(|f| f.primary_key)
2902 .map(|f| f.name)
2903 .unwrap_or("id");
2904
2905 let mut child_soft_delete = false;
2906 let mut m2m_junction: Option<String> = None;
2907
2908 let resolved = if let Some(spec) = rev_spec {
2909 child_soft_delete = spec.soft_delete;
2910 Ok((
2911 spec.target_table.to_string(),
2912 spec.fk_column.to_string(),
2913 T::TABLE.to_string(),
2914 pk.to_string(),
2915 ))
2916 } else if let Some(spec) = m2m_spec {
2917 // M2M count: junction table = "<parent>_<field>", columns
2918 // parent_id / child_id. The subquery counts junction rows.
2919 m2m_junction = Some(format!("{}_{}", T::TABLE, spec.field_name));
2920 // child_table / fk_column are unused for the M2M shape, but
2921 // the tuple still carries parent_table + parent_pk for the
2922 // correlation in build_query_for.
2923 Ok((
2924 spec.target_table.to_string(),
2925 "child_id".to_string(),
2926 T::TABLE.to_string(),
2927 pk.to_string(),
2928 ))
2929 } else {
2930 // No DECLARED relation matched. Fall back to auto-discovery
2931 // (gaps2 #45): scan the model registry for any child whose
2932 // FK points back at this parent's table, and match `relation`
2933 // against the conventional name forms. Declared relations
2934 // always win above; this only runs as a fallback.
2935 match discover_reverse_relation::<T>(relation) {
2936 AutoDiscovery::Resolved {
2937 child_table,
2938 fk_column,
2939 soft_delete,
2940 } => {
2941 child_soft_delete = soft_delete;
2942 Ok((child_table, fk_column, T::TABLE.to_string(), pk.to_string()))
2943 }
2944 AutoDiscovery::Ambiguous(candidates) => Err(format!(
2945 "umbral::orm::annotate_related: ambiguous reverse relation `{relation}` on `{}` — candidates: [{}]; declare a `#[umbral(reverse_fk = \"<fk>\")] ReverseSet<Child>` field to disambiguate",
2946 T::NAME,
2947 candidates.join(", "),
2948 )),
2949 AutoDiscovery::NotFound(discoverable) => {
2950 let declared = T::REVERSE_FK_RELATIONS
2951 .iter()
2952 .map(|r| r.field_name)
2953 .collect::<Vec<_>>()
2954 .join(", ");
2955 let m2m = T::M2M_RELATIONS
2956 .iter()
2957 .map(|r| r.field_name)
2958 .collect::<Vec<_>>()
2959 .join(", ");
2960 Err(format!(
2961 "umbral::orm::annotate_related: `{relation}` is not a reverse-FK or M2M relation on `{}` — reverse-FK relations: [{declared}], M2M relations: [{m2m}], auto-discoverable children: [{}]",
2962 T::NAME,
2963 discoverable.join(", "),
2964 ))
2965 }
2966 }
2967 };
2968
2969 self.annotations.push(RelatedAnnotation {
2970 alias: alias.to_string(),
2971 agg,
2972 resolved,
2973 child_soft_delete,
2974 child_filter: None,
2975 m2m_junction,
2976 });
2977 self
2978 }
2979
2980 /// Sugar for the overwhelmingly common annotation:
2981 /// `annotate_related("<relation>_count", relation, Aggregate::count())`.
2982 /// `.annotate_count("comment_set")` exposes the value under the
2983 /// `comment_set_count` alias in [`Self::fetch_annotated`].
2984 /// `ORDER BY <annotation alias>` — sort by a value you annotated, not by a
2985 /// column (gaps3 #29).
2986 ///
2987 /// This is what a leaderboard needs and what the ORM could not express:
2988 /// `order_by` is typed against the model's own columns, so it cannot name a
2989 /// computed alias, and without this there was no way to ask for "the top 20
2990 /// authors *by post count*". A live consumer worked around it by pulling
2991 /// whole tables into a `HashMap` and sorting in Rust — the aggregation engine
2992 /// was already there, it just wasn't reachable.
2993 ///
2994 /// ```ignore
2995 /// AuthUser::objects()
2996 /// .annotate_count("post_set")
2997 /// .order_by_annotation("post_set__count", true) // desc
2998 /// .limit(20)
2999 /// .fetch_annotated()
3000 /// .await?
3001 /// ```
3002 ///
3003 /// An alias that was never annotated is a loud error at query time, not
3004 /// silently-wrong SQL — see [`Self::check_annotations`].
3005 pub fn order_by_annotation(mut self, alias: &str, desc: bool) -> Self {
3006 self.annotation_order.push((alias.to_string(), desc));
3007 self.explicit_order = true;
3008 self
3009 }
3010
3011 pub fn annotate_count(self, relation: &str) -> Self {
3012 let alias = format!("{relation}_count");
3013 self.annotate_related(&alias, relation, crate::orm::Aggregate::count())
3014 }
3015
3016 /// Like [`Self::annotate_count`] but counts only the children
3017 /// matching `pred` — a filtered count over `"comments"`.
3018 /// `C` is the CHILD model, so the predicate is typed against the
3019 /// child's columns (`comment::MODERATION.eq("visible")`). The
3020 /// predicate renders into the correlated count subquery's WHERE
3021 /// alongside the FK correlation and the auto soft-delete filter.
3022 ///
3023 /// ```rust,ignore
3024 /// Plugin::objects()
3025 /// .annotate_count_where::<PluginComment>(
3026 /// "visible_comments",
3027 /// "comment_set",
3028 /// plugin_comment::MODERATION.eq("visible"),
3029 /// )
3030 /// ```
3031 pub fn annotate_count_where<C: crate::orm::Model>(
3032 self,
3033 alias: &str,
3034 relation: &str,
3035 pred: crate::orm::Predicate<C>,
3036 ) -> Self {
3037 // Render the child predicate to a backend-default SimpleExpr.
3038 // The count subquery embeds one expression; the equality /
3039 // comparison predicates used for child filters render the same
3040 // on both backends, so the default `cond` is correct.
3041 let child_filter = pred.cond_for("postgres");
3042 let mut queryset = self.annotate_related(alias, relation, crate::orm::Aggregate::count());
3043 // The just-pushed annotation is the last one; attach the filter.
3044 if let Some(last) = queryset.annotations.last_mut() {
3045 last.child_filter = Some(child_filter);
3046 }
3047 queryset
3048 }
3049
3050 /// Loud-failure check for poisoned annotations (unknown relation
3051 /// names recorded by the infallible builder). Called by every
3052 /// fallible consumer before SQL runs.
3053 fn check_annotations(&self) -> Result<(), sqlx::Error> {
3054 for ann in &self.annotations {
3055 if let Err(msg) = &ann.resolved {
3056 return Err(sqlx::Error::Protocol(msg.clone()));
3057 }
3058 }
3059 // gaps3 #29: `order_by_annotation("typo")` must fail loudly here, not
3060 // emit SQL that orders by a column the database has never heard of (or,
3061 // worse, silently matches a real column and returns confidently wrong
3062 // rows).
3063 for (alias, _) in &self.annotation_order {
3064 if !self.annotations.iter().any(|a| &a.alias == alias) {
3065 let known: Vec<&str> = self.annotations.iter().map(|a| a.alias.as_str()).collect();
3066 return Err(sqlx::Error::Protocol(format!(
3067 "order_by_annotation(\"{alias}\") names an annotation that was never added; \
3068 annotated aliases on this queryset: {known:?}"
3069 )));
3070 }
3071 }
3072 Ok(())
3073 }
3074
3075 /// [`Self::fetch_annotated`], deserialized into one struct of your own
3076 /// (gaps3 #29).
3077 ///
3078 /// `fetch_annotated` hands back `(T, Map<String, JsonValue>)`, so a handler
3079 /// that wants a flat row still writes `map["post_count"].as_i64().unwrap_or(0)`
3080 /// per field. That JSON-poking is plausibly *why* a live consumer skipped the
3081 /// aggregation engine entirely and rebuilt GROUP BY with `HashMap`s in Rust —
3082 /// the feature existed but wasn't reachable from a handler.
3083 ///
3084 /// The model's own columns and its annotations are merged into one object and
3085 /// deserialized, so the target struct just names what it wants:
3086 ///
3087 /// ```ignore
3088 /// #[derive(Deserialize)]
3089 /// struct Leader { id: i64, username: String, post_set__count: i64 }
3090 ///
3091 /// let top: Vec<Leader> = AuthUser::objects()
3092 /// .annotate_count("post_set")
3093 /// .order_by_annotation("post_set__count", true)
3094 /// .limit(20)
3095 /// .fetch_annotated_as::<Leader>()
3096 /// .await?;
3097 /// ```
3098 ///
3099 /// A field the query didn't produce is a deserialize error, not a silent
3100 /// default — a typo in the struct should not read as "zero posts".
3101 pub async fn fetch_annotated_as<R>(self) -> Result<Vec<R>, sqlx::Error>
3102 where
3103 R: serde::de::DeserializeOwned,
3104 T: serde::Serialize
3105 + for<'r> sqlx::FromRow<'r, sqlx::sqlite::SqliteRow>
3106 + for<'r> sqlx::FromRow<'r, sqlx::postgres::PgRow>,
3107 {
3108 let rows = self.fetch_annotated().await?;
3109 let mut out = Vec::with_capacity(rows.len());
3110 for (model, annotations) in rows {
3111 let mut obj = match serde_json::to_value(&model) {
3112 Ok(JsonValue::Object(o)) => o,
3113 _ => serde_json::Map::new(),
3114 };
3115 // Annotations win on a name clash: you asked for the computed value.
3116 for (k, v) in annotations {
3117 obj.insert(k, v);
3118 }
3119 out.push(
3120 serde_json::from_value::<R>(JsonValue::Object(obj))
3121 .map_err(|e| sqlx::Error::Protocol(format!("fetch_annotated_as: {e}")))?,
3122 );
3123 }
3124 Ok(out)
3125 }
3126
3127 /// Run the SELECT and return every matching row **with its
3128 /// annotation values** — the execution terminal for
3129 /// [`Self::annotate_related`] / [`Self::annotate_count`]. One
3130 /// query; each row's annotations arrive as an `alias → JSON
3131 /// value` map (count → integer, AVG → float/null, SUM/MAX/MIN →
3132 /// typed per the child column, NULL on empty sets for the
3133 /// non-count aggregates).
3134 pub async fn fetch_annotated(
3135 self,
3136 ) -> Result<Vec<(T, serde_json::Map<String, JsonValue>)>, sqlx::Error>
3137 where
3138 T: for<'r> sqlx::FromRow<'r, sqlx::sqlite::SqliteRow>
3139 + for<'r> sqlx::FromRow<'r, sqlx::postgres::PgRow>,
3140 {
3141 self.check_annotations()?;
3142 // Child-column types for SUM/MAX/MIN decoding, resolved from
3143 // the runtime registry (the child model is known by table
3144 // name only). `None` falls back to decode_agg's string path.
3145 let source_types: Vec<(String, crate::orm::Aggregate, Option<crate::orm::SqlType>)> = {
3146 let registry_up = crate::migrate::is_initialised();
3147 self.annotations
3148 .iter()
3149 .map(|ann| {
3150 let ty = match (&ann.resolved, registry_up, ann.agg.source_column()) {
3151 (Ok((child_table, ..)), true, Some(col)) => {
3152 crate::migrate::registered_models()
3153 .into_iter()
3154 .find(|m| m.table == *child_table)
3155 .and_then(|m| m.fields.iter().find(|f| f.name == col).map(|f| f.ty))
3156 }
3157 _ => None,
3158 };
3159 (ann.alias.clone(), ann.agg.clone(), ty)
3160 })
3161 .collect()
3162 };
3163
3164 let pool = resolve_pool::<T>(self.explicit_pool.clone(), crate::db::RouteOp::Read);
3165 match pool {
3166 DbPool::Sqlite(pool) => {
3167 let q = self.build_query_for("sqlite");
3168 let (sql, vals) = q.build_sqlx(SqliteQueryBuilder);
3169 let rows = sqlx::query_with::<sqlx::Sqlite, _>(&sql, vals)
3170 .fetch_all(&pool)
3171 .await?;
3172 let mut out = Vec::with_capacity(rows.len());
3173 for row in &rows {
3174 let t = <T as sqlx::FromRow<_>>::from_row(row)?;
3175 let mut anns = serde_json::Map::with_capacity(source_types.len());
3176 for (alias, agg, ty) in &source_types {
3177 anns.insert(
3178 alias.clone(),
3179 backend_sqlite::decode_agg(row, alias, agg, *ty)?,
3180 );
3181 }
3182 out.push((t, anns));
3183 }
3184 Ok(out)
3185 }
3186 DbPool::Postgres(pool) => {
3187 let q = self.build_query_for("postgres");
3188 let (sql, vals) = q.build_sqlx(PostgresQueryBuilder);
3189 let rows = sqlx::query_with::<sqlx::Postgres, _>(&sql, vals)
3190 .fetch_all(&pool)
3191 .await?;
3192 let mut out = Vec::with_capacity(rows.len());
3193 for row in &rows {
3194 let t = <T as sqlx::FromRow<_>>::from_row(row)?;
3195 let mut anns = serde_json::Map::with_capacity(source_types.len());
3196 for (alias, agg, ty) in &source_types {
3197 anns.insert(alias.clone(), backend_pg::decode_agg(row, alias, agg, *ty)?);
3198 }
3199 out.push((t, anns));
3200 }
3201 Ok(out)
3202 }
3203 }
3204 }
3205
3206 /// `DELETE FROM table WHERE <predicates>`. Returns the number of
3207 /// rows deleted. With no `.filter` calls, deletes every row.
3208 ///
3209 /// Fires `bulk_post_delete:<table>` once with the list of removed
3210 /// PKs when at least one row was deleted. Per-row `pre_delete` /
3211 /// `post_delete` are NOT fired by this path — use
3212 /// [`Manager::delete_instance`] when per-row signal semantics are
3213 /// required.
3214 ///
3215 /// Feature #72: for `#[umbral(soft_delete)]` models this rewrites
3216 /// to `UPDATE ... SET deleted_at = NOW() WHERE ...` so rows
3217 /// survive in the DB (filtered out of subsequent queries by the
3218 /// auto `WHERE deleted_at IS NULL`). Call `.hard_delete()`
3219 /// beforehand for a real DELETE (GDPR purge, test cleanup).
3220 pub async fn delete(self) -> Result<u64, sqlx::Error> {
3221 // features #73: a view is read-only. Refuse before any SQL is built, so the
3222 // error names the model instead of surfacing as the driver's opaque
3223 // "cannot modify a view".
3224 if T::VIEW.is_some() {
3225 return Err(sqlx::Error::Protocol(
3226 crate::orm::write::WriteError::ReadOnlyView {
3227 table: T::TABLE.to_string(),
3228 }
3229 .to_string(),
3230 ));
3231 }
3232 // Feature #72 — soft-delete redirect. The whole `delete()`
3233 // contract collapses to an UPDATE setting `deleted_at`. We
3234 // keep the bulk_post_delete signal so subscribers see the
3235 // same event shape regardless of the underlying SQL.
3236 if self.soft_delete_active && !self.hard_delete {
3237 return self.soft_delete_update().await;
3238 }
3239 let atomic = self.should_atomic_wrap();
3240 let pool = resolve_pool::<T>(self.explicit_pool.clone(), crate::db::RouteOp::Write);
3241 let backend = pool.backend_name();
3242 // gaps3 #54: the pre-image, read before the write lands. Free unless audited.
3243 let audit_before = self.audit_pre(backend).await;
3244 let mut stmt = self.build_delete_for(backend);
3245 let pk = pk_field::<T>();
3246 if let Some(field) = pk {
3247 stmt.returning_col(Alias::new(field.name));
3248 }
3249 let ids: Vec<JsonValue> = match pool {
3250 DbPool::Sqlite(pool) => {
3251 let (sql, values) = stmt.build_sqlx(SqliteQueryBuilder);
3252 let rows = if atomic {
3253 let mut tx = pool.begin().await?;
3254 let r = sqlx::query_with::<sqlx::Sqlite, _>(&sql, values)
3255 .fetch_all(&mut *tx)
3256 .await;
3257 match r {
3258 Ok(rows) => {
3259 tx.commit().await?;
3260 rows
3261 }
3262 Err(e) => {
3263 let _ = tx.rollback().await;
3264 return Err(e);
3265 }
3266 }
3267 } else {
3268 sqlx::query_with::<sqlx::Sqlite, _>(&sql, values)
3269 .fetch_all(&pool)
3270 .await?
3271 };
3272 match pk {
3273 Some(field) => rows
3274 .iter()
3275 .map(|r| backend_sqlite::pk_to_json(r, field.name, field.ty))
3276 .collect::<Result<_, _>>()?,
3277 None => Vec::new(),
3278 }
3279 }
3280 DbPool::Postgres(pool) => {
3281 let (sql, values) = stmt.build_sqlx(PostgresQueryBuilder);
3282 let rows = if atomic {
3283 let mut tx = pool.begin().await?;
3284 let r = sqlx::query_with::<sqlx::Postgres, _>(&sql, values)
3285 .fetch_all(&mut *tx)
3286 .await;
3287 match r {
3288 Ok(rows) => {
3289 tx.commit().await?;
3290 rows
3291 }
3292 Err(e) => {
3293 let _ = tx.rollback().await;
3294 return Err(e);
3295 }
3296 }
3297 } else {
3298 sqlx::query_with::<sqlx::Postgres, _>(&sql, values)
3299 .fetch_all(&pool)
3300 .await?
3301 };
3302 match pk {
3303 Some(field) => rows
3304 .iter()
3305 .map(|r| backend_pg::pk_to_json(r, field.name, field.ty))
3306 .collect::<Result<_, _>>()?,
3307 None => Vec::new(),
3308 }
3309 }
3310 };
3311 let count = ids.len() as u64;
3312 if !ids.is_empty() {
3313 Self::audit_post(audit_before, &ids, crate::orm::audit::DELETE).await;
3314
3315 // gaps3 #29: also emit the PER-ROW `post_delete:<table>`, not only the
3316 // bulk signal. `save()` and `update_or_create()` emit per-row signals;
3317 // `delete()` emitted only the bulk one, so `RealtimePlugin::on_model`
3318 // — which subscribes per-row — never saw a delete. A live consumer
3319 // hand-pushed a realtime event after every delete because of this.
3320 //
3321 // The payload is the primary key, not the whole row: the row is gone
3322 // by now, and re-reading it beforehand would cost a SELECT on every
3323 // delete in the app whether or not anything is listening. The id is
3324 // what a delete event is *for* — invalidate that row — and it is
3325 // exactly what realtime's default projection carries anyway.
3326 if let Some(pk) = pk_field::<T>() {
3327 for id in &ids {
3328 let payload = serde_json::json!({ pk.name: id });
3329 crate::signals::emit_post_delete_by_table(T::TABLE, payload).await;
3330 }
3331 }
3332 crate::signals::emit_bulk_post_delete::<T>(ids).await;
3333 }
3334 Ok(count)
3335 }
3336
3337 /// `UPDATE table SET col = <expr> WHERE <predicates>` using an
3338 /// F-expression for the new value.
3339 ///
3340 /// This is the companion to `update_values` for atomic column
3341 /// arithmetic. `F::col("views").add(1)` produces an [`FExpr`] that
3342 /// renders as `SET views = views + 1` — the database computes the
3343 /// increment atomically on the server side rather than needing a
3344 /// read-modify-write round-trip in application code.
3345 ///
3346 /// ```rust,ignore
3347 /// use umbral::orm::F;
3348 ///
3349 /// Post::objects()
3350 /// .filter(post::ID.eq(42))
3351 /// .update_expr("views", F::col("views").add(1))
3352 /// .await?;
3353 /// ```
3354 ///
3355 /// Mixing `update_values` and `update_expr` for different columns in
3356 /// one statement requires two separate calls. A combined API (a map
3357 /// where values can be either JSON or FExpr) would require a new sum
3358 /// type; deferred until a consumer surfaces the need.
3359 pub async fn update_expr(
3360 self,
3361 col_name: &str,
3362 expr: FExpr,
3363 ) -> Result<u64, crate::orm::write::WriteError> {
3364 use crate::orm::write::WriteError;
3365 // features #73: a view is read-only. Refuse before any SQL is built, so the
3366 // error names the model instead of surfacing as the driver's opaque
3367 // "cannot modify a view".
3368 if T::VIEW.is_some() {
3369 return Err(crate::orm::write::WriteError::ReadOnlyView {
3370 table: T::TABLE.to_string(),
3371 });
3372 }
3373 // Validate the column exists on the model.
3374 let field = T::FIELDS
3375 .iter()
3376 .find(|f| f.name == col_name)
3377 .ok_or_else(|| WriteError::UnknownColumn {
3378 field: col_name.to_string(),
3379 })?;
3380 if field.primary_key {
3381 // Silently skip PK rewrites, same as update_values.
3382 return Ok(0);
3383 }
3384 let pool = resolve_pool::<T>(self.explicit_pool.clone(), crate::db::RouteOp::Write);
3385 let backend = pool.backend_name();
3386 // gaps3 #54: the pre-image, read before the write lands. Free unless audited.
3387 let audit_before = self.audit_pre(backend).await;
3388
3389 let mut stmt = sea_query::Query::update();
3390 stmt.table(crate::db::router::schema_qualified_table(T::TABLE));
3391 stmt.value(Alias::new(field.name), expr.to_simple_expr());
3392 for p in &self.predicates {
3393 stmt.and_where(p.cond_for(backend));
3394 }
3395 // Route the implicit visibility filter through the single seam
3396 // (`implicit_predicates`) instead of a 4th hand-inlined copy of the
3397 // soft-delete guard — so a future implicit predicate (row-level
3398 // tenant scope) reaches update_expr too, not just reads/update_values.
3399 for e in self.implicit_predicates() {
3400 stmt.and_where(e);
3401 }
3402 let pk = pk_field::<T>();
3403 if let Some(pkf) = pk {
3404 stmt.returning_col(Alias::new(pkf.name));
3405 }
3406
3407 let ids: Vec<JsonValue> = match pool {
3408 DbPool::Sqlite(pool) => {
3409 let (sql, values) = stmt.build_sqlx(SqliteQueryBuilder);
3410 let rows = sqlx::query_with::<sqlx::Sqlite, _>(&sql, values)
3411 .fetch_all(&pool)
3412 .await?;
3413 match pk {
3414 Some(pkf) => rows
3415 .iter()
3416 .map(|r| backend_sqlite::pk_to_json(r, pkf.name, pkf.ty))
3417 .collect::<Result<_, _>>()
3418 .map_err(crate::orm::write::WriteError::Sqlx)?,
3419 None => Vec::new(),
3420 }
3421 }
3422 DbPool::Postgres(pool) => {
3423 let (sql, values) = stmt.build_sqlx(PostgresQueryBuilder);
3424 let rows = sqlx::query_with::<sqlx::Postgres, _>(&sql, values)
3425 .fetch_all(&pool)
3426 .await?;
3427 match pk {
3428 Some(pkf) => rows
3429 .iter()
3430 .map(|r| backend_pg::pk_to_json(r, pkf.name, pkf.ty))
3431 .collect::<Result<_, _>>()
3432 .map_err(crate::orm::write::WriteError::Sqlx)?,
3433 None => Vec::new(),
3434 }
3435 }
3436 };
3437 let count = ids.len() as u64;
3438 if !ids.is_empty() {
3439 Self::audit_post(audit_before, &ids, crate::orm::audit::UPDATE).await;
3440 crate::signals::emit_bulk_post_save::<T>(ids, false).await;
3441 }
3442 Ok(count)
3443 }
3444
3445 /// `UPDATE table SET k=v[, ...] WHERE <predicates>`. The values
3446 /// map provides `column_name → JSON value` pairs; each is
3447 /// converted to a `sea_query::Value` per the column's declared
3448 /// `SqlType` via [`crate::orm::write::json_to_sea_value`]. Returns
3449 /// the number of rows affected.
3450 ///
3451 /// Unknown columns in the map fail loudly with
3452 /// `WriteError::UnknownColumn`. JSON `null` is rejected for
3453 /// non-nullable columns; supplying a column that exists but is
3454 /// absent from the map is silently a no-op (the column keeps its
3455 /// current value — PATCH semantics, not PUT).
3456 ///
3457 /// Fires `bulk_post_save:<table>` once with `{ ids, created:
3458 /// false, actor }` when at least one row matched. Per-row
3459 /// `pre_save` / `post_save` are NOT fired — use [`Manager::save`]
3460 /// when per-row signal semantics are required.
3461 pub async fn update_values(
3462 self,
3463 values: serde_json::Map<String, serde_json::Value>,
3464 ) -> Result<u64, crate::orm::write::WriteError> {
3465 // features #73: a view is read-only. Refuse before any SQL is built, so the
3466 // error names the model instead of surfacing as the driver's opaque
3467 // "cannot modify a view".
3468 if T::VIEW.is_some() {
3469 return Err(crate::orm::write::WriteError::ReadOnlyView {
3470 table: T::TABLE.to_string(),
3471 });
3472 }
3473 let atomic = self.should_atomic_wrap();
3474 let pool = resolve_pool::<T>(self.explicit_pool.clone(), crate::db::RouteOp::Write);
3475 let backend = pool.backend_name();
3476 // gaps3 #54: the pre-image, read before the write lands. Free unless audited.
3477 let audit_before = self.audit_pre(backend).await;
3478 let mut stmt = self.build_update_for(backend, &values)?;
3479 // RETURNING <pk> so bulk_post_save can include the matched ids.
3480 let pk = pk_field::<T>();
3481 if let Some(field) = pk {
3482 stmt.returning_col(Alias::new(field.name));
3483 }
3484 let ids: Vec<JsonValue> = match pool {
3485 DbPool::Sqlite(pool) => {
3486 let (sql, values) = stmt.build_sqlx(SqliteQueryBuilder);
3487 let rows = if atomic {
3488 let mut tx = pool
3489 .begin()
3490 .await
3491 .map_err(crate::orm::write::WriteError::Sqlx)?;
3492 let r = sqlx::query_with::<sqlx::Sqlite, _>(&sql, values)
3493 .fetch_all(&mut *tx)
3494 .await;
3495 match r {
3496 Ok(rows) => {
3497 tx.commit()
3498 .await
3499 .map_err(crate::orm::write::WriteError::Sqlx)?;
3500 rows
3501 }
3502 Err(e) => {
3503 let _ = tx.rollback().await;
3504 return Err(crate::orm::write::WriteError::Sqlx(e));
3505 }
3506 }
3507 } else {
3508 sqlx::query_with::<sqlx::Sqlite, _>(&sql, values)
3509 .fetch_all(&pool)
3510 .await?
3511 };
3512 match pk {
3513 Some(field) => rows
3514 .iter()
3515 .map(|r| backend_sqlite::pk_to_json(r, field.name, field.ty))
3516 .collect::<Result<_, _>>()
3517 .map_err(crate::orm::write::WriteError::Sqlx)?,
3518 None => Vec::new(),
3519 }
3520 }
3521 DbPool::Postgres(pool) => {
3522 let (sql, values) = stmt.build_sqlx(PostgresQueryBuilder);
3523 let rows = if atomic {
3524 let mut tx = pool
3525 .begin()
3526 .await
3527 .map_err(crate::orm::write::WriteError::Sqlx)?;
3528 let r = sqlx::query_with::<sqlx::Postgres, _>(&sql, values)
3529 .fetch_all(&mut *tx)
3530 .await;
3531 match r {
3532 Ok(rows) => {
3533 tx.commit()
3534 .await
3535 .map_err(crate::orm::write::WriteError::Sqlx)?;
3536 rows
3537 }
3538 Err(e) => {
3539 let _ = tx.rollback().await;
3540 return Err(crate::orm::write::WriteError::Sqlx(e));
3541 }
3542 }
3543 } else {
3544 sqlx::query_with::<sqlx::Postgres, _>(&sql, values)
3545 .fetch_all(&pool)
3546 .await?
3547 };
3548 match pk {
3549 Some(field) => rows
3550 .iter()
3551 .map(|r| backend_pg::pk_to_json(r, field.name, field.ty))
3552 .collect::<Result<_, _>>()
3553 .map_err(crate::orm::write::WriteError::Sqlx)?,
3554 None => Vec::new(),
3555 }
3556 }
3557 };
3558 let count = ids.len() as u64;
3559 if !ids.is_empty() {
3560 Self::audit_post(audit_before, &ids, crate::orm::audit::UPDATE).await;
3561 crate::signals::emit_bulk_post_save::<T>(ids, false).await;
3562 }
3563 Ok(count)
3564 }
3565
3566 /// Helper: build the DELETE statement for the active backend.
3567 /// Public-by-virtue-of-being-pub(crate) so the `_pg` and (future)
3568 /// `_sqlite` explicit-pool variants can share the SQL builder.
3569 fn build_delete_for(&self, backend_name: &str) -> sea_query::DeleteStatement {
3570 let mut stmt = Query::delete();
3571 stmt.from_table(crate::db::router::schema_qualified_table(T::TABLE));
3572 for p in &self.predicates {
3573 stmt.and_where(p.cond_for(backend_name));
3574 }
3575 stmt
3576 }
3577
3578 /// Feature #72 — soft-delete rewrite: turn `DELETE FROM table
3579 /// WHERE ...` into `UPDATE table SET deleted_at = NOW() WHERE
3580 /// ... AND deleted_at IS NULL`. The trailing `IS NULL` guard
3581 /// makes the operation idempotent: re-soft-deleting an already-
3582 /// soft-deleted row doesn't bump its timestamp. Fires
3583 /// `bulk_post_delete:<table>` so subscribers see the same event
3584 /// shape as a hard delete.
3585 async fn soft_delete_update(self) -> Result<u64, sqlx::Error> {
3586 let pool = resolve_pool::<T>(self.explicit_pool.clone(), crate::db::RouteOp::Write);
3587 let backend = pool.backend_name();
3588 // gaps3 #54: the pre-image, read before the write lands. Free unless audited.
3589 let audit_before = self.audit_pre(backend).await;
3590 let now = chrono::Utc::now();
3591 let mut stmt = sea_query::Query::update();
3592 stmt.table(crate::db::router::schema_qualified_table(T::TABLE));
3593 stmt.value(
3594 Alias::new("deleted_at"),
3595 sea_query::Value::ChronoDateTimeUtc(Some(Box::new(now))),
3596 );
3597 for p in &self.predicates {
3598 stmt.and_where(p.cond_for(backend));
3599 }
3600 // Idempotency guard — never bump an already-set deleted_at.
3601 stmt.and_where(sea_query::Expr::col(Alias::new("deleted_at")).is_null());
3602 let pk = pk_field::<T>();
3603 if let Some(pkf) = pk {
3604 stmt.returning_col(Alias::new(pkf.name));
3605 }
3606 // The cascade must find the children BEFORE the parent is stamped: it
3607 // locates them through the parent's still-matching live-rows predicate
3608 // (same predicates + `deleted_at IS NULL` guard as the UPDATE above).
3609 let meta = crate::migrate::ModelMeta::for_::<T>();
3610 let cascade_sel = pk.map(|pkf| {
3611 let mut sel = sea_query::Query::select();
3612 sel.column(Alias::new(pkf.name))
3613 .from(crate::db::router::schema_qualified_table(T::TABLE));
3614 for p in &self.predicates {
3615 sel.and_where(p.cond_for(backend));
3616 }
3617 sel.and_where(sea_query::Expr::col(Alias::new("deleted_at")).is_null());
3618 sel
3619 });
3620
3621 // Parent + cascade in ONE transaction, ALWAYS (gaps3 #53) — not gated on
3622 // `should_atomic_wrap`. A half-applied cascade leaves exactly the orphaned
3623 // live children this fix exists to prevent.
3624 let ids: Vec<JsonValue> = match pool {
3625 DbPool::Sqlite(pool) => {
3626 let (sql, values) = stmt.build_sqlx(SqliteQueryBuilder);
3627 let mut tx = pool.begin().await?;
3628 if let Some(sel) = cascade_sel {
3629 let mut conn = crate::orm::soft_delete_cascade::CascadeConn::Sqlite(&mut tx);
3630 crate::orm::soft_delete_cascade::cascade_soft_delete(
3631 &mut conn, &meta, sel, now,
3632 )
3633 .await?;
3634 }
3635 let rows = sqlx::query_with::<sqlx::Sqlite, _>(&sql, values)
3636 .fetch_all(&mut *tx)
3637 .await?;
3638 tx.commit().await?;
3639 match pk {
3640 Some(field) => rows
3641 .iter()
3642 .map(|r| backend_sqlite::pk_to_json(r, field.name, field.ty))
3643 .collect::<Result<_, _>>()?,
3644 None => Vec::new(),
3645 }
3646 }
3647 DbPool::Postgres(pool) => {
3648 let (sql, values) = stmt.build_sqlx(PostgresQueryBuilder);
3649 let mut tx = pool.begin().await?;
3650 if let Some(sel) = cascade_sel {
3651 let mut conn = crate::orm::soft_delete_cascade::CascadeConn::Pg(&mut tx);
3652 crate::orm::soft_delete_cascade::cascade_soft_delete(
3653 &mut conn, &meta, sel, now,
3654 )
3655 .await?;
3656 }
3657 let rows = sqlx::query_with::<sqlx::Postgres, _>(&sql, values)
3658 .fetch_all(&mut *tx)
3659 .await?;
3660 tx.commit().await?;
3661 match pk {
3662 Some(field) => rows
3663 .iter()
3664 .map(|r| backend_pg::pk_to_json(r, field.name, field.ty))
3665 .collect::<Result<_, _>>()?,
3666 None => Vec::new(),
3667 }
3668 }
3669 };
3670 let count = ids.len() as u64;
3671 if !ids.is_empty() {
3672 Self::audit_post(audit_before, &ids, crate::orm::audit::DELETE).await;
3673 crate::signals::emit_bulk_post_delete::<T>(ids).await;
3674 }
3675 Ok(count)
3676 }
3677
3678 /// Helper: build the UPDATE statement for the active backend.
3679 /// Walks the `values` map, validates each column against the
3680 /// model's `FIELDS` metadata, converts the JSON value via
3681 /// `write::json_to_sea_value`, and threads the accumulated
3682 /// predicates into the WHERE clause.
3683 fn build_update_for(
3684 &self,
3685 backend_name: &str,
3686 values: &serde_json::Map<String, serde_json::Value>,
3687 ) -> Result<sea_query::UpdateStatement, crate::orm::write::WriteError> {
3688 use crate::orm::write::{WriteError, json_to_sea_value};
3689 let mut stmt = Query::update();
3690 stmt.table(crate::db::router::schema_qualified_table(T::TABLE));
3691 for (col_name, val) in values {
3692 // Look up the column on the model. Unknown column names
3693 // fail loudly here rather than producing a bad UPDATE.
3694 let field = T::FIELDS
3695 .iter()
3696 .find(|f| f.name == col_name.as_str())
3697 .ok_or_else(|| WriteError::UnknownColumn {
3698 field: col_name.clone(),
3699 })?;
3700 // Reject attempts to overwrite the PK via update_values.
3701 // The QuerySet's WHERE clause is the only way to identify
3702 // rows; rewriting the PK while filtering on the old one
3703 // is a footgun.
3704 if field.primary_key {
3705 continue;
3706 }
3707 // features #83: an UPDATE must normalize too — otherwise a field
3708 // arrives clean on create and dirty on edit.
3709 let normalized = crate::orm::write::normalize_json(field.trim, field.lowercase, val);
3710 let val = normalized.as_ref().unwrap_or(val);
3711 let sea_value =
3712 json_to_sea_value(field.ty, val, field.nullable, field.name, fk_pk_hint(field))?;
3713 stmt.value(Alias::new(field.name), sea_value);
3714 }
3715 for p in &self.predicates {
3716 stmt.and_where(p.cond_for(backend_name));
3717 }
3718 for implicit in self.implicit_predicates() {
3719 stmt.and_where(implicit);
3720 }
3721 Ok(stmt)
3722 }
3723
3724 /// Run the SELECT against an explicit `PgPool` and return every
3725 /// matching row. Bound by `FromRow<PgRow>` alone so models with
3726 /// Postgres-only field types compile.
3727 pub async fn fetch_pg(self, pool: &sqlx::PgPool) -> Result<Vec<T>, sqlx::Error>
3728 where
3729 T: for<'r> sqlx::FromRow<'r, sqlx::postgres::PgRow>,
3730 {
3731 // gaps4 #24: refuse a chained hydration feature this low-level terminal
3732 // would silently drop, rather than hand back un-hydrated rows.
3733 guard_pg_terminal_unsupported(
3734 "fetch_pg",
3735 &self.select_related,
3736 &self.prefetch_related,
3737 &self.join_related,
3738 &self.only_cols,
3739 )?;
3740 let q = self.build_query_for("postgres");
3741 let (sql, values) = q.build_sqlx(PostgresQueryBuilder);
3742 sqlx::query_as_with::<sqlx::Postgres, T, _>(&sql, values)
3743 .fetch_all(pool)
3744 .await
3745 }
3746
3747 /// Run the SELECT against an explicit `PgPool` with LIMIT 1.
3748 pub async fn first_pg(mut self, pool: &sqlx::PgPool) -> Result<Option<T>, sqlx::Error>
3749 where
3750 T: for<'r> sqlx::FromRow<'r, sqlx::postgres::PgRow>,
3751 {
3752 guard_pg_terminal_unsupported(
3753 "first_pg",
3754 &self.select_related,
3755 &self.prefetch_related,
3756 &self.join_related,
3757 &self.only_cols,
3758 )?;
3759 self.query.limit(1);
3760 let q = self.build_query_for("postgres");
3761 let (sql, values) = q.build_sqlx(PostgresQueryBuilder);
3762 sqlx::query_as_with::<sqlx::Postgres, T, _>(&sql, values)
3763 .fetch_optional(pool)
3764 .await
3765 }
3766
3767 /// Run `SELECT COUNT(*)` against an explicit `PgPool`. No FromRow
3768 /// bound on `T` — the count tuple type is `(i64,)`.
3769 pub async fn count_pg(self, pool: &sqlx::PgPool) -> Result<i64, sqlx::Error> {
3770 let mut rebuilt = self.build_query_for("postgres");
3771 rebuilt.clear_selects();
3772 rebuilt.expr(Func::count(Expr::col(sea_query::Asterisk)));
3773 rebuilt.reset_limit();
3774 rebuilt.reset_offset();
3775 let (sql, values) = rebuilt.build_sqlx(PostgresQueryBuilder);
3776 let (n,): (i64,) = sqlx::query_as_with::<sqlx::Postgres, (i64,), _>(&sql, values)
3777 .fetch_one(pool)
3778 .await?;
3779 Ok(n)
3780 }
3781
3782 /// Return whether any row matches, against an explicit `PgPool`.
3783 pub async fn exists_pg(self, pool: &sqlx::PgPool) -> Result<bool, sqlx::Error>
3784 where
3785 T: for<'r> sqlx::FromRow<'r, sqlx::postgres::PgRow>,
3786 {
3787 let rows = self.limit(1).fetch_pg(pool).await?;
3788 Ok(!rows.is_empty())
3789 }
3790
3791 /// Exactly-one terminal against an explicit `PgPool`.
3792 /// See [`QuerySet::get`] for the error-variant semantics.
3793 pub async fn get_pg(self, pool: &sqlx::PgPool) -> Result<T, GetError>
3794 where
3795 T: for<'r> sqlx::FromRow<'r, sqlx::postgres::PgRow>,
3796 {
3797 let mut rows = self.limit(2).fetch_pg(pool).await.map_err(GetError::Sqlx)?;
3798 match rows.len() {
3799 0 => Err(GetError::NotFound),
3800 1 => Ok(rows.pop().unwrap()),
3801 _ => Err(GetError::MultipleObjectsReturned),
3802 }
3803 }
3804}
3805
3806/// Delegating chainable + terminal surface on `Manager<T>`.
3807///
3808/// Lets users write `Post::objects().filter(...).fetch().await` without
3809/// a separate `.query()` hop. Each method constructs the initial
3810/// `SelectStatement` against `T::TABLE` with one column per
3811/// `T::FIELDS` entry, wraps it in a fresh `QuerySet<T>`, and forwards.
3812impl<T: Model> Manager<T> {
3813 fn queryset(&self) -> QuerySet<T> {
3814 let columns: Vec<Alias> = T::FIELDS.iter().map(|f| Alias::new(f.name)).collect();
3815 let query = Query::select()
3816 .columns(columns)
3817 .from(crate::db::router::schema_qualified_table(T::TABLE))
3818 .take();
3819 let mut qs = QuerySet::new(query);
3820 // BUG-8: seed the default ORDER BY from `Model::ORDERING` so
3821 // terminals that don't see an explicit `.order_by(...)` still
3822 // get a deterministic row order.
3823 qs.default_ordering = T::ORDERING.to_vec();
3824 // Propagate the Manager's atomic override so QuerySet
3825 // terminals inherit it without the caller re-specifying.
3826 qs.atomic = self.atomic;
3827 // Feature #72 — snapshot the model's soft-delete opt-in into
3828 // the QuerySet so the build_query_for path knows whether to
3829 // auto-inject `WHERE deleted_at IS NULL`. Without this
3830 // snapshot the `impl<T> QuerySet<T>` path can't read
3831 // `T::SOFT_DELETE` (T is unbounded there).
3832 qs.soft_delete_active = T::SOFT_DELETE;
3833 qs
3834 }
3835
3836 /// Resolve whether write terminals on this Manager should auto-wrap
3837 /// in a transaction. Per-call override > builder global.
3838 fn should_atomic_wrap(&self) -> bool {
3839 self.atomic.unwrap_or_else(crate::db::atomic_default)
3840 }
3841
3842 /// See `QuerySet::filter`.
3843 pub fn filter(&self, p: Predicate<T>) -> QuerySet<T> {
3844 self.queryset().filter(p)
3845 }
3846
3847 /// See `QuerySet::exclude`.
3848 pub fn exclude(&self, p: Predicate<T>) -> QuerySet<T> {
3849 self.queryset().exclude(p)
3850 }
3851
3852 /// A bare [`QuerySet`] over every row — the `Model::objects().all()` form.
3853 ///
3854 /// The entry point when you need a `QuerySet` terminal without a
3855 /// filter: a grouped aggregate over the whole table, an unfiltered
3856 /// `aggregate`, or just an explicit "all rows" for readability.
3857 pub fn all(&self) -> QuerySet<T> {
3858 self.queryset()
3859 }
3860
3861 /// See [`QuerySet::aggregate`] — single-row aggregate over every row.
3862 ///
3863 /// Forwards from the manager so `Model::objects().aggregate(...)`
3864 /// works without an intervening `.filter(...)` / `.on(...)`.
3865 pub async fn aggregate(
3866 &self,
3867 aggs: &[(&str, crate::orm::Aggregate)],
3868 ) -> Result<JsonValue, sqlx::Error> {
3869 self.queryset().aggregate(aggs).await
3870 }
3871
3872 /// See [`QuerySet::annotate`] — grouped aggregate (`GROUP BY <group_cols>`).
3873 ///
3874 /// Forwards from the manager so the documented
3875 /// `Model::objects().annotate(&["status"], &[("count", Aggregate::count())])`
3876 /// (a grouped count over `"status"`) compiles
3877 /// directly, without a filter first.
3878 pub async fn annotate(
3879 &self,
3880 group_cols: &[&str],
3881 aggs: &[(&str, crate::orm::Aggregate)],
3882 ) -> Result<Vec<JsonValue>, sqlx::Error> {
3883 self.queryset().annotate(group_cols, aggs).await
3884 }
3885
3886 /// Feature #72 — see `QuerySet::with_deleted`.
3887 pub fn with_deleted(&self) -> QuerySet<T> {
3888 self.queryset().with_deleted()
3889 }
3890
3891 /// Feature #72 — see `QuerySet::only_deleted`.
3892 pub fn only_deleted(&self) -> QuerySet<T> {
3893 self.queryset().only_deleted()
3894 }
3895
3896 /// Gap #111 — see [`QuerySet::only`].
3897 pub fn only(&self, cols: &[&str]) -> QuerySet<T> {
3898 self.queryset().only(cols)
3899 }
3900
3901 /// See [`QuerySet::join_related`].
3902 pub fn join_related(&self, field_name: impl Into<String>) -> QuerySet<T> {
3903 self.queryset().join_related(field_name)
3904 }
3905
3906 /// See [`QuerySet::join_related_many`].
3907 pub fn join_related_many(&self, field_names: &[&str]) -> QuerySet<T> {
3908 self.queryset().join_related_many(field_names)
3909 }
3910
3911 /// See [`QuerySet::left_join_related`].
3912 pub fn left_join_related(&self, path: impl Into<String>) -> QuerySet<T> {
3913 self.queryset().left_join_related(path)
3914 }
3915
3916 /// See [`QuerySet::inner_join_related`].
3917 pub fn inner_join_related(&self, path: impl Into<String>) -> QuerySet<T> {
3918 self.queryset().inner_join_related(path)
3919 }
3920
3921 /// See [`QuerySet::right_join_related`].
3922 pub fn right_join_related(&self, path: impl Into<String>) -> QuerySet<T> {
3923 self.queryset().right_join_related(path)
3924 }
3925
3926 /// See [`QuerySet::select_related`].
3927 pub fn select_related(&self, field_name: impl Into<String>) -> QuerySet<T> {
3928 self.queryset().select_related(field_name)
3929 }
3930
3931 /// See [`QuerySet::select_related_many`].
3932 pub fn select_related_many(&self, field_names: &[&str]) -> QuerySet<T> {
3933 self.queryset().select_related_many(field_names)
3934 }
3935
3936 /// See [`QuerySet::prefetch_related`].
3937 pub fn prefetch_related(&self, field_name: impl Into<String>) -> QuerySet<T> {
3938 self.queryset().prefetch_related(field_name)
3939 }
3940
3941 /// See [`QuerySet::prefetch_related_many`].
3942 pub fn prefetch_related_many(&self, field_names: &[&str]) -> QuerySet<T> {
3943 self.queryset().prefetch_related_many(field_names)
3944 }
3945
3946 /// Feature #72 — see `QuerySet::hard_delete`.
3947 pub fn hard_delete(&self) -> QuerySet<T> {
3948 self.queryset().hard_delete()
3949 }
3950
3951 /// See `QuerySet::into_subquery`.
3952 pub fn into_subquery(&self, col_name: &str) -> crate::orm::Subquery {
3953 self.queryset().into_subquery(col_name)
3954 }
3955
3956 /// See `QuerySet::order_by`.
3957 pub fn order_by(&self, o: OrderExpr<T>) -> QuerySet<T> {
3958 self.queryset().order_by(o)
3959 }
3960
3961 /// See `QuerySet::limit`.
3962 pub fn limit(&self, n: u64) -> QuerySet<T> {
3963 self.queryset().limit(n)
3964 }
3965
3966 /// See `QuerySet::offset`.
3967 pub fn offset(&self, n: u64) -> QuerySet<T> {
3968 self.queryset().offset(n)
3969 }
3970
3971 /// See `QuerySet::on`.
3972 pub fn on(&self, pool: &sqlx::SqlitePool) -> QuerySet<T> {
3973 self.queryset().on(pool)
3974 }
3975
3976 /// See `QuerySet::on_pg`.
3977 pub fn on_pg(&self, pool: &sqlx::PgPool) -> QuerySet<T> {
3978 self.queryset().on_pg(pool)
3979 }
3980
3981 /// See `QuerySet::fetch`.
3982 pub async fn fetch(&self) -> Result<Vec<T>, sqlx::Error>
3983 where
3984 T: for<'r> sqlx::FromRow<'r, sqlx::sqlite::SqliteRow>
3985 + for<'r> sqlx::FromRow<'r, sqlx::postgres::PgRow>
3986 + HydrateRelated,
3987 {
3988 self.queryset().fetch().await
3989 }
3990
3991 /// See `QuerySet::first`.
3992 pub async fn first(&self) -> Result<Option<T>, sqlx::Error>
3993 where
3994 T: for<'r> sqlx::FromRow<'r, sqlx::sqlite::SqliteRow>
3995 + for<'r> sqlx::FromRow<'r, sqlx::postgres::PgRow>
3996 + HydrateRelated,
3997 {
3998 self.queryset().first().await
3999 }
4000
4001 /// See `QuerySet::count`.
4002 pub async fn count(&self) -> Result<i64, sqlx::Error> {
4003 self.queryset().count().await
4004 }
4005
4006 /// See [`QuerySet::annotate_related`] — starts an annotated chain
4007 /// from the manager, like `filter` does.
4008 pub fn annotate_related(
4009 &self,
4010 alias: &str,
4011 relation: &str,
4012 agg: crate::orm::Aggregate,
4013 ) -> QuerySet<T> {
4014 self.queryset().annotate_related(alias, relation, agg)
4015 }
4016
4017 /// See [`QuerySet::annotate_count`].
4018 pub fn annotate_count(&self, relation: &str) -> QuerySet<T> {
4019 self.queryset().annotate_count(relation)
4020 }
4021
4022 /// See [`QuerySet::order_by_annotation`].
4023 pub fn order_by_annotation(&self, alias: &str, desc: bool) -> QuerySet<T> {
4024 self.queryset().order_by_annotation(alias, desc)
4025 }
4026
4027 /// See [`QuerySet::annotate_as`] — the typed GROUP BY rollup.
4028 pub async fn annotate_as<R>(
4029 &self,
4030 group_cols: &[&str],
4031 aggs: &[(&str, crate::orm::Aggregate)],
4032 ) -> Result<Vec<R>, sqlx::Error>
4033 where
4034 R: serde::de::DeserializeOwned,
4035 {
4036 self.queryset().annotate_as::<R>(group_cols, aggs).await
4037 }
4038
4039 /// See [`QuerySet::annotate_count_where`] — starts a filtered
4040 /// annotated chain from the manager.
4041 pub fn annotate_count_where<C: crate::orm::Model>(
4042 &self,
4043 alias: &str,
4044 relation: &str,
4045 pred: crate::orm::Predicate<C>,
4046 ) -> QuerySet<T> {
4047 self.queryset()
4048 .annotate_count_where::<C>(alias, relation, pred)
4049 }
4050
4051 /// See [`QuerySet::fetch_annotated`].
4052 pub async fn fetch_annotated(
4053 &self,
4054 ) -> Result<Vec<(T, serde_json::Map<String, JsonValue>)>, sqlx::Error>
4055 where
4056 T: for<'r> sqlx::FromRow<'r, sqlx::sqlite::SqliteRow>
4057 + for<'r> sqlx::FromRow<'r, sqlx::postgres::PgRow>,
4058 {
4059 self.queryset().fetch_annotated().await
4060 }
4061
4062 /// See [`QuerySet::values`].
4063 pub async fn values(&self, columns: &[&str]) -> Result<Vec<JsonValue>, sqlx::Error> {
4064 self.queryset().values(columns).await
4065 }
4066
4067 /// See `QuerySet::exists`.
4068 pub async fn exists(&self) -> Result<bool, sqlx::Error>
4069 where
4070 T: for<'r> sqlx::FromRow<'r, sqlx::sqlite::SqliteRow>
4071 + for<'r> sqlx::FromRow<'r, sqlx::postgres::PgRow>
4072 + HydrateRelated,
4073 {
4074 self.queryset().exists().await
4075 }
4076
4077 /// `.get(predicate)` — sugar for `.filter(predicate).get()`.
4078 ///
4079 /// The one-liner: `User::objects().get(user::ID.eq(1))`.
4080 /// See [`QuerySet::get`] for error-variant semantics.
4081 pub async fn get(&self, p: Predicate<T>) -> Result<T, GetError>
4082 where
4083 T: for<'r> sqlx::FromRow<'r, sqlx::sqlite::SqliteRow>
4084 + for<'r> sqlx::FromRow<'r, sqlx::postgres::PgRow>
4085 + HydrateRelated,
4086 {
4087 self.queryset().filter(p).get().await
4088 }
4089
4090 /// See [`QuerySet::fetch_pg`].
4091 pub async fn fetch_pg(&self, pool: &sqlx::PgPool) -> Result<Vec<T>, sqlx::Error>
4092 where
4093 T: for<'r> sqlx::FromRow<'r, sqlx::postgres::PgRow>,
4094 {
4095 self.queryset().fetch_pg(pool).await
4096 }
4097
4098 /// See [`QuerySet::first_pg`].
4099 pub async fn first_pg(&self, pool: &sqlx::PgPool) -> Result<Option<T>, sqlx::Error>
4100 where
4101 T: for<'r> sqlx::FromRow<'r, sqlx::postgres::PgRow>,
4102 {
4103 self.queryset().first_pg(pool).await
4104 }
4105
4106 /// See [`QuerySet::count_pg`].
4107 pub async fn count_pg(&self, pool: &sqlx::PgPool) -> Result<i64, sqlx::Error> {
4108 self.queryset().count_pg(pool).await
4109 }
4110
4111 /// See [`QuerySet::exists_pg`].
4112 pub async fn exists_pg(&self, pool: &sqlx::PgPool) -> Result<bool, sqlx::Error>
4113 where
4114 T: for<'r> sqlx::FromRow<'r, sqlx::postgres::PgRow>,
4115 {
4116 self.queryset().exists_pg(pool).await
4117 }
4118
4119 /// Postgres-only sugar for `.filter(predicate).get_pg(pool)`.
4120 pub async fn get_pg(&self, pool: &sqlx::PgPool, p: Predicate<T>) -> Result<T, GetError>
4121 where
4122 T: for<'r> sqlx::FromRow<'r, sqlx::postgres::PgRow>,
4123 {
4124 self.queryset().filter(p).get_pg(pool).await
4125 }
4126
4127 // =====================================================================
4128 // Write methods — INSERT.
4129 //
4130 // `create(instance)` does one row; `bulk_create([...])` does many in
4131 // a single multi-VALUES INSERT. Both serialise the instance(s) to a
4132 // JSON map via `serde::Serialize`, look up each field in the model's
4133 // `FIELDS` metadata, and bind values through
4134 // [`crate::orm::write::json_to_sea_value`].
4135 //
4136 // PK handling:
4137 // - Default value (0 for ints, nil for UUIDs, empty for String):
4138 // omitted from the INSERT column list so the DB autoincrement /
4139 // default kicks in.
4140 // - Explicit non-default value: included in the INSERT so the
4141 // caller can supply UUIDs / slug PKs themselves.
4142 // =====================================================================
4143
4144 /// INSERT one row, return the row as it now exists in the
4145 /// database (with any autoincrement PK populated). Uses the
4146 /// ambient pool via `Manager::queryset().resolve_pool`.
4147 pub async fn create(&self, mut instance: T) -> Result<T, crate::orm::write::WriteError>
4148 where
4149 T: serde::Serialize
4150 + for<'r> sqlx::FromRow<'r, sqlx::sqlite::SqliteRow>
4151 + for<'r> sqlx::FromRow<'r, sqlx::postgres::PgRow>
4152 + HydrateRelated,
4153 {
4154 use crate::orm::write::WriteError;
4155 // features #73: a view is read-only. Refuse before any SQL is built, so the
4156 // error names the model instead of surfacing as the driver's opaque
4157 // "cannot modify a view".
4158 if T::VIEW.is_some() {
4159 return Err(crate::orm::write::WriteError::ReadOnlyView {
4160 table: T::TABLE.to_string(),
4161 });
4162 }
4163 let map = serialize_to_map(&instance)?;
4164
4165 // Same pre-DB validation pipeline the dynamic
4166 // `insert_json` path runs — choices + FK existence +
4167 // M2M shape. Empty-string + required-field checks are
4168 // intentionally relaxed on the typed path: a Rust
4169 // `pub title: String` field set to `""` is the caller's
4170 // deliberate choice, not a form-default leak, and
4171 // missing-required can't happen because the struct
4172 // forced the caller to supply every column at compile
4173 // time. We only validate the things the typed path
4174 // can't catch at compile time.
4175 let meta = crate::migrate::ModelMeta::for_::<T>();
4176 let validation_errors = crate::orm::validation::validate_on_typed_create(&meta, &map).await;
4177 if !validation_errors.is_empty() {
4178 return Err(WriteError::Multiple {
4179 errors: validation_errors,
4180 });
4181 }
4182
4183 let pool = resolve_pool::<T>(None, crate::db::RouteOp::Write);
4184 let backend = pool.backend_name();
4185 let stmt = build_insert_one_for::<T>(backend, &map)?;
4186 let atomic = self.should_atomic_wrap();
4187 // Post-execution SQL classification: turns the DB's
4188 // UNIQUE / FK / NOT NULL / CHECK violations into the
4189 // structured `WriteError` variants instead of a raw
4190 // `Sqlx(_)` 500. Symmetric with `DynQuerySet::insert_json`.
4191 match pool {
4192 DbPool::Sqlite(pool) => {
4193 let (sql, values) = stmt.build_sqlx(SqliteQueryBuilder);
4194 let row_result = if atomic {
4195 let mut tx = pool.begin().await.map_err(WriteError::Sqlx)?;
4196 let r = sqlx::query_as_with::<sqlx::Sqlite, T, _>(&sql, values)
4197 .fetch_one(&mut *tx)
4198 .await;
4199 match r {
4200 Ok(row) => {
4201 tx.commit().await.map_err(WriteError::Sqlx)?;
4202 Ok(row)
4203 }
4204 Err(e) => {
4205 let _ = tx.rollback().await;
4206 Err(e)
4207 }
4208 }
4209 } else {
4210 sqlx::query_as_with::<sqlx::Sqlite, T, _>(&sql, values)
4211 .fetch_one(&pool)
4212 .await
4213 };
4214 let mut row = row_result.map_err(|e| {
4215 crate::orm::validation::classify_sql_error(&e, &map)
4216 .unwrap_or(WriteError::Sqlx(e))
4217 })?;
4218 // BUG-16 step 2: every materialised row, including the
4219 // post-INSERT readback, needs `parent_id` +
4220 // `junction_table` seeded on its M2M slots — otherwise
4221 // `row.tags.add(...)` is a silent no-op.
4222 row.set_m2m_parent_ids();
4223 // Carry form-staged M2M pending ids from the caller's
4224 // instance onto the readback row, then flush them to
4225 // junction rows now that parent_id + junction_table are
4226 // seeded on the readback row.
4227 instance.take_pending_m2m_into(&mut row);
4228 row.write_pending_m2m().await?;
4229 // gaps3 #29: `create()` emitted NOTHING, while `save()` and
4230 // `update_or_create()` both emit per-row `post_save`. So every
4231 // `.create()` was invisible to signals — and therefore to
4232 // `RealtimePlugin::on_model`, which is why a live consumer
4233 // hand-pushed a realtime event after all 13 of its writes.
4234 crate::signals::emit_post_save::<T>(&row, true).await;
4235 // gaps3 #54: the created row IS the after-image.
4236 crate::orm::audit::record(
4237 &crate::migrate::ModelMeta::for_::<T>(),
4238 &serde_json::to_value(&row)
4239 .ok()
4240 .and_then(|v| {
4241 v.as_object().map(|o| {
4242 crate::orm::audit::pk_of(&crate::migrate::ModelMeta::for_::<T>(), o)
4243 })
4244 })
4245 .unwrap_or_default(),
4246 crate::orm::audit::CREATE,
4247 None,
4248 serde_json::to_value(&row)
4249 .ok()
4250 .and_then(|v| v.as_object().cloned())
4251 .as_ref(),
4252 )
4253 .await;
4254 Ok(row)
4255 }
4256 DbPool::Postgres(pool) => {
4257 let (sql, values) = stmt.build_sqlx(PostgresQueryBuilder);
4258 let row_result = if atomic {
4259 let mut tx = pool.begin().await.map_err(WriteError::Sqlx)?;
4260 let r = sqlx::query_as_with::<sqlx::Postgres, T, _>(&sql, values)
4261 .fetch_one(&mut *tx)
4262 .await;
4263 match r {
4264 Ok(row) => {
4265 tx.commit().await.map_err(WriteError::Sqlx)?;
4266 Ok(row)
4267 }
4268 Err(e) => {
4269 let _ = tx.rollback().await;
4270 Err(e)
4271 }
4272 }
4273 } else {
4274 sqlx::query_as_with::<sqlx::Postgres, T, _>(&sql, values)
4275 .fetch_one(&pool)
4276 .await
4277 };
4278 let mut row = row_result.map_err(|e| {
4279 crate::orm::validation::classify_sql_error(&e, &map)
4280 .unwrap_or(WriteError::Sqlx(e))
4281 })?;
4282 row.set_m2m_parent_ids();
4283 instance.take_pending_m2m_into(&mut row);
4284 row.write_pending_m2m().await?;
4285 // gaps3 #29: `create()` emitted NOTHING, while `save()` and
4286 // `update_or_create()` both emit per-row `post_save`. So every
4287 // `.create()` was invisible to signals — and therefore to
4288 // `RealtimePlugin::on_model`, which is why a live consumer
4289 // hand-pushed a realtime event after all 13 of its writes.
4290 crate::signals::emit_post_save::<T>(&row, true).await;
4291 // gaps3 #54: the created row IS the after-image.
4292 crate::orm::audit::record(
4293 &crate::migrate::ModelMeta::for_::<T>(),
4294 &serde_json::to_value(&row)
4295 .ok()
4296 .and_then(|v| {
4297 v.as_object().map(|o| {
4298 crate::orm::audit::pk_of(&crate::migrate::ModelMeta::for_::<T>(), o)
4299 })
4300 })
4301 .unwrap_or_default(),
4302 crate::orm::audit::CREATE,
4303 None,
4304 serde_json::to_value(&row)
4305 .ok()
4306 .and_then(|v| v.as_object().cloned())
4307 .as_ref(),
4308 )
4309 .await;
4310 Ok(row)
4311 }
4312 }
4313 }
4314
4315 /// INSERT many rows in a single statement. Returns the number of
4316 /// rows inserted. The full populated rows aren't materialised —
4317 /// use a follow-up `Model::objects().filter(...).fetch()` if you
4318 /// need them.
4319 ///
4320 /// Empty input is a no-op (returns Ok(0)) — the alternative
4321 /// (building an `INSERT INTO t () VALUES ()` and failing at the
4322 /// DB) doesn't help anyone.
4323 ///
4324 /// Fires `bulk_post_save:<table>` once with `{ ids, created: true,
4325 /// actor }` when at least one row was inserted. Per-row
4326 /// `pre_save` / `post_save` are NOT fired — use [`Self::save`]
4327 /// when per-row signal semantics are required.
4328 pub async fn bulk_create(&self, instances: Vec<T>) -> Result<u64, crate::orm::write::WriteError>
4329 where
4330 T: serde::Serialize,
4331 {
4332 use crate::orm::write::WriteError;
4333 // features #73: a view is read-only. Refuse before any SQL is built, so the
4334 // error names the model instead of surfacing as the driver's opaque
4335 // "cannot modify a view".
4336 if T::VIEW.is_some() {
4337 return Err(crate::orm::write::WriteError::ReadOnlyView {
4338 table: T::TABLE.to_string(),
4339 });
4340 }
4341 if instances.is_empty() {
4342 return Ok(0);
4343 }
4344 let maps: Result<Vec<_>, _> = instances.iter().map(serialize_to_map).collect();
4345 let maps = maps?;
4346 // Validate every instance through the typed-create
4347 // pipeline. Collected into one `Multiple` so a caller
4348 // that submitted ten rows and got two bad ones can fix
4349 // both in one pass.
4350 let meta = crate::migrate::ModelMeta::for_::<T>();
4351 let mut all_errors: Vec<WriteError> = Vec::new();
4352 for map in &maps {
4353 let errs = crate::orm::validation::validate_on_typed_create(&meta, map).await;
4354 all_errors.extend(errs);
4355 }
4356 if !all_errors.is_empty() {
4357 return Err(WriteError::Multiple { errors: all_errors });
4358 }
4359 let pool = resolve_pool::<T>(None, crate::db::RouteOp::Write);
4360 let backend = pool.backend_name();
4361 let mut stmt = build_insert_many_for::<T>(backend, &maps)?;
4362 // First row's map is used to enrich UNIQUE / FK
4363 // messages with the offending value when the engine
4364 // doesn't name it. Imperfect for bulk (the failing row
4365 // could be later in the batch) but better than the raw
4366 // sqlx error.
4367 let first_map = maps.first().cloned().unwrap_or_default();
4368 // Add `RETURNING <pk>` so the bulk_post_save signal payload can
4369 // carry the inserted PKs. Both backends support this — SQLite
4370 // since 3.35, Postgres natively. Replaces the previous
4371 // `execute()` + rows_affected path; count comes from the
4372 // returned row vector instead.
4373 let pk = pk_field::<T>();
4374 if let Some(field) = pk {
4375 stmt.returning_col(Alias::new(field.name));
4376 }
4377 let atomic = self.should_atomic_wrap();
4378 let ids: Vec<JsonValue> = match pool {
4379 DbPool::Sqlite(pool) => {
4380 let (sql, values) = stmt.build_sqlx(SqliteQueryBuilder);
4381 let rows = if atomic {
4382 let mut tx = pool.begin().await.map_err(WriteError::Sqlx)?;
4383 let r = sqlx::query_with::<sqlx::Sqlite, _>(&sql, values)
4384 .fetch_all(&mut *tx)
4385 .await;
4386 match r {
4387 Ok(rows) => {
4388 tx.commit().await.map_err(WriteError::Sqlx)?;
4389 rows
4390 }
4391 Err(e) => {
4392 let _ = tx.rollback().await;
4393 return Err(crate::orm::validation::classify_sql_error(&e, &first_map)
4394 .unwrap_or(WriteError::Sqlx(e)));
4395 }
4396 }
4397 } else {
4398 sqlx::query_with::<sqlx::Sqlite, _>(&sql, values)
4399 .fetch_all(&pool)
4400 .await
4401 .map_err(|e| {
4402 crate::orm::validation::classify_sql_error(&e, &first_map)
4403 .unwrap_or(WriteError::Sqlx(e))
4404 })?
4405 };
4406 match pk {
4407 Some(field) => rows
4408 .iter()
4409 .map(|r| backend_sqlite::pk_to_json(r, field.name, field.ty))
4410 .collect::<Result<_, _>>()
4411 .map_err(WriteError::Sqlx)?,
4412 None => Vec::new(),
4413 }
4414 }
4415 DbPool::Postgres(pool) => {
4416 let (sql, values) = stmt.build_sqlx(PostgresQueryBuilder);
4417 let rows = if atomic {
4418 let mut tx = pool.begin().await.map_err(WriteError::Sqlx)?;
4419 let r = sqlx::query_with::<sqlx::Postgres, _>(&sql, values)
4420 .fetch_all(&mut *tx)
4421 .await;
4422 match r {
4423 Ok(rows) => {
4424 tx.commit().await.map_err(WriteError::Sqlx)?;
4425 rows
4426 }
4427 Err(e) => {
4428 let _ = tx.rollback().await;
4429 return Err(crate::orm::validation::classify_sql_error(&e, &first_map)
4430 .unwrap_or(WriteError::Sqlx(e)));
4431 }
4432 }
4433 } else {
4434 sqlx::query_with::<sqlx::Postgres, _>(&sql, values)
4435 .fetch_all(&pool)
4436 .await
4437 .map_err(|e| {
4438 crate::orm::validation::classify_sql_error(&e, &first_map)
4439 .unwrap_or(WriteError::Sqlx(e))
4440 })?
4441 };
4442 match pk {
4443 Some(field) => rows
4444 .iter()
4445 .map(|r| backend_pg::pk_to_json(r, field.name, field.ty))
4446 .collect::<Result<_, _>>()
4447 .map_err(WriteError::Sqlx)?,
4448 None => Vec::new(),
4449 }
4450 }
4451 };
4452 let count = ids.len() as u64;
4453 if !ids.is_empty() {
4454 QuerySet::<T>::audit_post(Vec::new(), &ids, crate::orm::audit::CREATE).await;
4455 crate::signals::emit_bulk_post_save::<T>(ids, true).await;
4456 }
4457 Ok(count)
4458 }
4459
4460 /// The `get_or_create` terminal: fetch the first row matching `predicate`;
4461 /// if none exists, insert `defaults` and return it. Returns
4462 /// `(row, created)` so the caller can branch on whether the write
4463 /// happened. Two queries on the miss path (filter+first then create),
4464 /// one query on the hit path.
4465 ///
4466 /// ## Concurrency
4467 ///
4468 /// Convergent under concurrent callers: if two callers both miss the
4469 /// SELECT and race to INSERT, the one that loses gets a
4470 /// `UniqueViolation`; that error is caught here and the existing row
4471 /// is re-fetched, so both callers return the same row with
4472 /// `created = false` for the loser. A UNIQUE constraint on the
4473 /// predicate columns is required for true at-most-one semantics — the
4474 /// constraint is what makes the convergence deterministic.
4475 pub async fn get_or_create(
4476 &self,
4477 predicate: Predicate<T>,
4478 defaults: T,
4479 ) -> Result<(T, bool), crate::orm::write::WriteError>
4480 where
4481 T: serde::Serialize
4482 + for<'r> sqlx::FromRow<'r, sqlx::sqlite::SqliteRow>
4483 + for<'r> sqlx::FromRow<'r, sqlx::postgres::PgRow>
4484 + HydrateRelated,
4485 {
4486 use crate::orm::write::WriteError;
4487
4488 // Read-your-writes: probe for the existing row on the WRITE database,
4489 // not a (possibly lagging) read replica — otherwise a read/write-split
4490 // router could miss a just-written row and insert a duplicate. The
4491 // following `create()` already resolves the same write target.
4492 let write_pool = resolve_pool::<T>(None, crate::db::RouteOp::Write);
4493 if let Some(existing) = pin_to_pool(self.filter(predicate.clone()), &write_pool)
4494 .first()
4495 .await
4496 .map_err(WriteError::Sqlx)?
4497 {
4498 return Ok((existing, false));
4499 }
4500
4501 // Attempt the INSERT. On a UNIQUE violation (concurrent writer won the
4502 // race between our SELECT and this INSERT), catch the error and
4503 // re-SELECT to return the now-existing row with created=false.
4504 // This is the standard try-insert-then-fetch convergence pattern.
4505 //
4506 // Note on transaction semantics: a plain INSERT is atomic at the
4507 // statement level. Wrapping SELECT+INSERT in a serialisable transaction
4508 // would be stricter but requires SAVEPOINT support to recover from the
4509 // Postgres "aborted transaction" state after a constraint violation —
4510 // a per-operation SAVEPOINT would add two extra round-trips on every
4511 // write for marginal gain. The UNIQUE-constraint backstop plus this
4512 // re-fetch gives the same observable guarantee: callers always converge
4513 // on the same row and never see a spurious UniqueViolation.
4514 match self.create(defaults).await {
4515 Ok(created) => Ok((created, true)),
4516 Err(WriteError::UniqueViolation { .. }) => {
4517 // A concurrent writer inserted the row between our SELECT and
4518 // our INSERT. Re-fetch the now-existing row.
4519 let existing = pin_to_pool(self.filter(predicate), &write_pool)
4520 .first()
4521 .await
4522 .map_err(WriteError::Sqlx)?
4523 .ok_or_else(|| {
4524 WriteError::Sqlx(sqlx::Error::Protocol(
4525 "get_or_create: row vanished after UniqueViolation re-fetch"
4526 .to_string(),
4527 ))
4528 })?;
4529 Ok((existing, false))
4530 }
4531 Err(e) => Err(e),
4532 }
4533 }
4534
4535 /// The `update_or_create` terminal: fetch the first row matching
4536 /// `predicate`; if found, update its non-PK columns with the
4537 /// `defaults` instance's values and return the fresh row;
4538 /// otherwise insert `defaults` and return it. Returns
4539 /// `(row, created)` so the caller can branch on the path taken.
4540 ///
4541 /// The defaults' PK is intentionally ignored on the update path —
4542 /// the matched row keeps its original PK. On the insert path the
4543 /// defaults' PK is honoured (autoincrement sentinel `0` → DB
4544 /// assigns; explicit value → DB uses it).
4545 ///
4546 /// ## Concurrency
4547 ///
4548 /// Convergent under concurrent callers: if two callers both miss the
4549 /// SELECT and race to INSERT, the loser gets a `UniqueViolation`; that
4550 /// error is caught here and the existing row is re-fetched, then the
4551 /// update is applied to it. Both callers converge on the same row, with
4552 /// `created = false` for the loser. A UNIQUE constraint on the predicate
4553 /// columns is required for deterministic convergence.
4554 ///
4555 /// Implementation: 2 queries on the hit path (`first` + UPDATE + re-fetch),
4556 /// 2 queries on the miss+create path (`first` + `create`), or
4557 /// 4 queries on the miss+race path (`first` + failed-INSERT + `first`
4558 /// + UPDATE + re-fetch).
4559 pub async fn update_or_create(
4560 &self,
4561 predicate: Predicate<T>,
4562 defaults: T,
4563 ) -> Result<(T, bool), crate::orm::write::WriteError>
4564 where
4565 T: serde::Serialize
4566 + Clone
4567 + for<'r> sqlx::FromRow<'r, sqlx::sqlite::SqliteRow>
4568 + for<'r> sqlx::FromRow<'r, sqlx::postgres::PgRow>
4569 + HydrateRelated,
4570 {
4571 use crate::orm::write::WriteError;
4572 let pk = pk_field::<T>().ok_or_else(|| {
4573 WriteError::Sqlx(sqlx::Error::Protocol(
4574 "update_or_create: model has no primary key".to_string(),
4575 ))
4576 })?;
4577 let pk_name = pk.name;
4578
4579 // Read-your-writes: the existence probe and the post-update re-fetch
4580 // run on the WRITE database, so a read/write-split router doesn't miss
4581 // a just-written row (duplicate insert) or read a stale row back. The
4582 // intervening UPDATE already routes to the same write target.
4583 let write_pool = resolve_pool::<T>(None, crate::db::RouteOp::Write);
4584
4585 // Shared helper: given the existing row (already fetched) and the
4586 // defaults instance, apply the non-PK column update and return the
4587 // re-fetched row. Used by both the direct-hit path and the
4588 // UniqueViolation-convergence path so the update logic is in one place.
4589 macro_rules! do_update {
4590 ($existing:expr, $defaults:expr) => {{
4591 let existing: T = $existing;
4592 let defaults: T = $defaults;
4593
4594 // Serialize defaults, drop the PK so the matched row's PK is
4595 // preserved, then UPDATE WHERE <pk_col> = <existing_pk>.
4596 let mut update_map = serialize_to_map(&defaults)?;
4597 update_map.remove(pk_name);
4598
4599 // Build a PK predicate from the existing row's serialized PK
4600 // value. Goes through serde_json so any PK type (i64, String,
4601 // Uuid) round-trips correctly through sea-query.
4602 let existing_json =
4603 serde_json::to_value(&existing).map_err(WriteError::SerializeFailed)?;
4604 let pk_value_json = existing_json
4605 .get(pk_name)
4606 .cloned()
4607 .unwrap_or(serde_json::Value::Null);
4608 let pk_sea = crate::orm::write::json_to_sea_value(
4609 pk.ty,
4610 &pk_value_json,
4611 false,
4612 pk_name,
4613 None,
4614 )?;
4615 let pk_pred: Predicate<T> =
4616 Predicate::new(sea_query::Expr::col(sea_query::Alias::new(pk_name)).eq(pk_sea));
4617
4618 // Run the UPDATE.
4619 self.filter(pk_pred).update_values(update_map).await?;
4620
4621 // Re-fetch to return the populated row.
4622 let pk_sea2 = crate::orm::write::json_to_sea_value(
4623 pk.ty,
4624 &pk_value_json,
4625 false,
4626 pk_name,
4627 None,
4628 )?;
4629 let refetch_pred: Predicate<T> = Predicate::new(
4630 sea_query::Expr::col(sea_query::Alias::new(pk_name)).eq(pk_sea2),
4631 );
4632 let updated_row: T = pin_to_pool(self.filter(refetch_pred), &write_pool)
4633 .first()
4634 .await
4635 .map_err(WriteError::Sqlx)?
4636 .ok_or_else(|| {
4637 WriteError::Sqlx(sqlx::Error::Protocol(
4638 "update_or_create: row vanished between UPDATE and re-fetch"
4639 .to_string(),
4640 ))
4641 })?;
4642
4643 // `update_values` above fires `bulk_post_save`; ALSO fire the
4644 // per-row `post_save` so signal / realtime `on_model` consumers
4645 // (which subscribe to the per-row event) see this upsert-update.
4646 // Without it, the CREATE branch (via `self.create()`) emits
4647 // `post_save` but the UPDATE branch was silent to those
4648 // consumers — an asymmetry that's very hard to reason about
4649 // from the call site (gaps3 #14). `bulk_post_save` and
4650 // `post_save` are distinct signal names, so no single consumer
4651 // fires twice.
4652 crate::signals::emit_post_save::<T>(&updated_row, false).await;
4653 updated_row
4654 }};
4655 }
4656
4657 if let Some(existing) = pin_to_pool(self.filter(predicate.clone()), &write_pool)
4658 .first()
4659 .await
4660 .map_err(WriteError::Sqlx)?
4661 {
4662 let updated = do_update!(existing, defaults);
4663 return Ok((updated, false));
4664 }
4665
4666 // Attempt the INSERT. On a UNIQUE violation (concurrent writer won the
4667 // race between our SELECT and this INSERT), catch the error, re-fetch
4668 // the now-existing row, and apply the update to it — same convergence
4669 // as get_or_create but with an extra UPDATE step.
4670 match self.create(defaults.clone()).await {
4671 Ok(created) => {
4672 // `create()` now fires `post_save` and records the audit row
4673 // itself (gaps3 #29), so this branch must NOT do either again —
4674 // it delegates to `create()` above. Before #29, `create()` was
4675 // signal-free and gaps3 #14 patched the gap here; that patch is
4676 // now a double-emit (and, since gaps3 #54, a double audit row).
4677 Ok((created, true))
4678 }
4679 Err(WriteError::UniqueViolation { .. }) => {
4680 // A concurrent writer inserted the row between our SELECT and
4681 // our INSERT. Re-fetch then update, same as the direct-hit path.
4682 let existing = pin_to_pool(self.filter(predicate), &write_pool)
4683 .first()
4684 .await
4685 .map_err(WriteError::Sqlx)?
4686 .ok_or_else(|| {
4687 WriteError::Sqlx(sqlx::Error::Protocol(
4688 "update_or_create: row vanished after UniqueViolation re-fetch"
4689 .to_string(),
4690 ))
4691 })?;
4692 let updated = do_update!(existing, defaults);
4693 Ok((updated, false))
4694 }
4695 Err(e) => Err(e),
4696 }
4697 }
4698
4699 /// INSERT-or-UPDATE keyed on the primary key. The row's PK column
4700 /// is the conflict target; on a hit, every non-PK column is
4701 /// overwritten with the supplied value. Returns the row as the DB
4702 /// stored it (post-upsert).
4703 ///
4704 /// Both backends use `INSERT ... ON CONFLICT(<pk>) DO UPDATE SET
4705 /// col = excluded.col, ...`. The SQLite and Postgres syntax happens
4706 /// to match exactly here so a single sea-query `OnConflict` builder
4707 /// covers both.
4708 pub async fn upsert(&self, instance: T) -> Result<T, crate::orm::write::WriteError>
4709 where
4710 T: serde::Serialize
4711 + for<'r> sqlx::FromRow<'r, sqlx::sqlite::SqliteRow>
4712 + for<'r> sqlx::FromRow<'r, sqlx::postgres::PgRow>,
4713 {
4714 let map = serialize_to_map(&instance)?;
4715 let pool = resolve_pool::<T>(None, crate::db::RouteOp::Write);
4716 let backend = pool.backend_name();
4717 let mut stmt = build_insert_one_for::<T>(backend, &map)?;
4718
4719 // Conflict target = PK column. update_columns = every non-PK
4720 // column the body included. sea-query renders `DO UPDATE SET
4721 // col = excluded.col` (SQLite) / `DO UPDATE SET col =
4722 // EXCLUDED.col` (PG) — both forms work cross-dialect.
4723 let pk_name = T::FIELDS
4724 .iter()
4725 .find(|f| f.primary_key)
4726 .map(|f| f.name)
4727 .ok_or_else(|| {
4728 crate::orm::write::WriteError::Sqlx(sqlx::Error::Protocol(
4729 "upsert: model has no primary key — use get_or_create or create instead"
4730 .to_string(),
4731 ))
4732 })?;
4733 let update_cols: Vec<Alias> = T::FIELDS
4734 .iter()
4735 .filter(|f| !f.primary_key && map.contains_key(f.name))
4736 .map(|f| Alias::new(f.name))
4737 .collect();
4738 let mut on_conflict = sea_query::OnConflict::column(Alias::new(pk_name));
4739 if !update_cols.is_empty() {
4740 on_conflict.update_columns(update_cols);
4741 } else {
4742 // No non-PK columns to overwrite — this is a "INSERT OR
4743 // IGNORE" shape. sea-query encodes that as `DO NOTHING`.
4744 on_conflict.do_nothing();
4745 }
4746 stmt.on_conflict(on_conflict);
4747
4748 match pool {
4749 DbPool::Sqlite(pool) => {
4750 let (sql, values) = stmt.build_sqlx(SqliteQueryBuilder);
4751 let row = sqlx::query_as_with::<sqlx::Sqlite, T, _>(&sql, values)
4752 .fetch_one(&pool)
4753 .await?;
4754 Ok(row)
4755 }
4756 DbPool::Postgres(pool) => {
4757 let (sql, values) = stmt.build_sqlx(PostgresQueryBuilder);
4758 let row = sqlx::query_as_with::<sqlx::Postgres, T, _>(&sql, values)
4759 .fetch_one(&pool)
4760 .await?;
4761 Ok(row)
4762 }
4763 }
4764 }
4765
4766 /// `create` against an explicit Postgres pool. The Postgres
4767 /// counterpart of [`Self::create`] for models with Postgres-only
4768 /// field types (Array, Inet, MacAddr, FullText), whose `FromRow`
4769 /// impl exists only for `PgRow`.
4770 ///
4771 /// **Low-level escape hatch (gaps4 #24).** Unlike [`Self::create`], this
4772 /// serializes + inserts + returns the row and NOTHING ELSE: it does not run
4773 /// validation, does not classify a DB error into a structured
4774 /// [`WriteError`], does not write form-staged **M2M** junctions, and does
4775 /// not fire `post_save` **signals** or record an **audit** entry. For a
4776 /// model with M2M relations, `#[umbral(audited)]`, or registered signal
4777 /// handlers, prefer the generic [`Self::create`] — it routes to Postgres
4778 /// through the ambient pool / `DatabaseRouter` and runs the full lifecycle.
4779 /// Reach for `create_pg` only for a Postgres-only-typed model that has none
4780 /// of those, or when you explicitly want the bare insert.
4781 pub async fn create_pg(
4782 &self,
4783 instance: T,
4784 pool: &sqlx::PgPool,
4785 ) -> Result<T, crate::orm::write::WriteError>
4786 where
4787 T: serde::Serialize + for<'r> sqlx::FromRow<'r, sqlx::postgres::PgRow>,
4788 {
4789 let map = serialize_to_map(&instance)?;
4790 let stmt = build_insert_one_for::<T>("postgres", &map)?;
4791 let (sql, values) = stmt.build_sqlx(PostgresQueryBuilder);
4792 let row = sqlx::query_as_with::<sqlx::Postgres, T, _>(&sql, values)
4793 .fetch_one(pool)
4794 .await?;
4795 Ok(row)
4796 }
4797
4798 /// Apply per-row differing values to a list of instances in one
4799 /// statement. Each instance carries its own PK and its own
4800 /// column values; the generated SQL uses one `CASE pk WHEN ...
4801 /// THEN ... END` per non-PK column, plus a `WHERE pk IN (...)`
4802 /// to scope the update.
4803 ///
4804 /// A `bulk_update(objs, fields)` that
4805 /// updates every non-PK column rather than asking the caller to
4806 /// list them. Returns the number of rows affected.
4807 ///
4808 /// Empty input is a no-op (returns 0). The pattern works on both
4809 /// SQLite and Postgres — the `CASE` expression is SQL-standard.
4810 ///
4811 /// Limitations:
4812 /// - All instances must have a non-default PK (the caller has
4813 /// already loaded the rows). Default-PK instances are skipped.
4814 /// - Bulk-write signals are NOT fired by this path — it's the
4815 /// `Manager::bulk_create` analogue for UPDATE, deliberately
4816 /// silent for speed.
4817 pub async fn bulk_update(&self, instances: Vec<T>) -> Result<u64, crate::orm::write::WriteError>
4818 where
4819 T: serde::Serialize,
4820 {
4821 use crate::orm::write::{WriteError, is_default_pk, json_to_sea_value};
4822 if instances.is_empty() {
4823 return Ok(0);
4824 }
4825 let pk = pk_field::<T>().ok_or_else(|| {
4826 WriteError::Sqlx(sqlx::Error::Protocol(
4827 "bulk_update: model has no primary key".to_string(),
4828 ))
4829 })?;
4830 let pk_name = pk.name;
4831 let pk_ty = pk.ty;
4832
4833 // Serialize every instance, collecting (pk_value, full_map)
4834 // for the CASE branches and the IN clause. Skip rows whose
4835 // PK is still the default sentinel — they were never
4836 // persisted and a bulk UPDATE on them is a no-op anyway.
4837 let mut serialized: Vec<(
4838 serde_json::Value,
4839 serde_json::Map<String, serde_json::Value>,
4840 )> = Vec::with_capacity(instances.len());
4841 for instance in &instances {
4842 let map = serialize_to_map(instance)?;
4843 let pk_val = map.get(pk_name).cloned().unwrap_or(serde_json::Value::Null);
4844 if is_default_pk(pk_ty, &pk_val) {
4845 continue;
4846 }
4847 serialized.push((pk_val, map));
4848 }
4849 if serialized.is_empty() {
4850 return Ok(0);
4851 }
4852
4853 // Collect the list of non-PK columns to update from the
4854 // first row (every row contributes the same column set
4855 // because they're typed instances of T).
4856 let update_cols: Vec<&crate::orm::FieldSpec> =
4857 T::FIELDS.iter().filter(|f| !f.primary_key).collect();
4858
4859 // Build the UPDATE: one CASE per column, IN clause for the
4860 // WHERE. Goes through sea-query's update statement for
4861 // backend portability.
4862 let mut stmt = sea_query::Query::update();
4863 stmt.table(crate::db::router::schema_qualified_table(T::TABLE));
4864
4865 for field in &update_cols {
4866 // CASE pk_col
4867 // WHEN <pk1> THEN <val1>
4868 // WHEN <pk2> THEN <val2>
4869 // ...
4870 // END
4871 let mut case = sea_query::CaseStatement::new();
4872 for (pk_val, map) in &serialized {
4873 let val = map
4874 .get(field.name)
4875 .cloned()
4876 .unwrap_or(serde_json::Value::Null);
4877 let cell = json_to_sea_value(
4878 field.ty,
4879 &val,
4880 field.nullable,
4881 field.name,
4882 fk_pk_hint(field),
4883 )?;
4884 let pk_sea = json_to_sea_value(pk_ty, pk_val, false, pk_name, None)?;
4885 case = case.case(sea_query::Expr::col(Alias::new(pk_name)).eq(pk_sea), cell);
4886 }
4887 stmt.value(Alias::new(field.name), case);
4888 }
4889
4890 // WHERE pk IN (<pk1>, <pk2>, ...)
4891 let pk_seas: Vec<sea_query::Value> = serialized
4892 .iter()
4893 .map(|(pk_val, _)| json_to_sea_value(pk_ty, pk_val, false, pk_name, None))
4894 .collect::<Result<_, _>>()?;
4895 stmt.and_where(sea_query::Expr::col(Alias::new(pk_name)).is_in(pk_seas));
4896
4897 let pool = resolve_pool::<T>(None, crate::db::RouteOp::Write);
4898 let affected = match pool {
4899 DbPool::Sqlite(pool) => {
4900 let (sql, values) = stmt.build_sqlx(SqliteQueryBuilder);
4901 sqlx::query_with::<sqlx::Sqlite, _>(&sql, values)
4902 .execute(&pool)
4903 .await
4904 .map_err(WriteError::Sqlx)?
4905 .rows_affected()
4906 }
4907 DbPool::Postgres(pool) => {
4908 let (sql, values) = stmt.build_sqlx(PostgresQueryBuilder);
4909 sqlx::query_with::<sqlx::Postgres, _>(&sql, values)
4910 .execute(&pool)
4911 .await
4912 .map_err(WriteError::Sqlx)?
4913 .rows_affected()
4914 }
4915 };
4916 Ok(affected)
4917 }
4918
4919 /// Run a hand-written SQL query and return typed `Vec<T>` rows.
4920 ///
4921 /// The escape hatch for queries the QuerySet builder can't (or
4922 /// shouldn't) model — CTEs, vendor-specific functions, ad-hoc
4923 /// reporting. Delegates to `sqlx::query_as` against the ambient
4924 /// pool and dispatches on backend, so user code stays portable.
4925 /// The string is sent verbatim; no parameter binding (use
4926 /// `Predicate` / the typed query path for parameterised
4927 /// queries). Inject input only after manual sanitisation.
4928 ///
4929 /// Skips the `select_related` / `prefetch_related` chain — those
4930 /// only apply to the typed QuerySet build path.
4931 pub async fn raw(&self, sql: &str) -> Result<Vec<T>, sqlx::Error>
4932 where
4933 T: for<'r> sqlx::FromRow<'r, sqlx::sqlite::SqliteRow>
4934 + for<'r> sqlx::FromRow<'r, sqlx::postgres::PgRow>,
4935 {
4936 // Routing: `raw` resolves to the READ database (most raw statements
4937 // are SELECTs, which this returns as `Vec<T>`). Under a read/write-
4938 // split router, a raw statement that WRITES must pin the write pool
4939 // explicitly via `.on(&pool)` / `.on_pg(&pool)`, since the router
4940 // cannot inspect arbitrary SQL to know it mutates.
4941 let pool = resolve_pool::<T>(None, crate::db::RouteOp::Read);
4942 match pool {
4943 DbPool::Sqlite(pool) => {
4944 sqlx::query_as::<sqlx::Sqlite, T>(sql)
4945 .fetch_all(&pool)
4946 .await
4947 }
4948 DbPool::Postgres(pool) => {
4949 sqlx::query_as::<sqlx::Postgres, T>(sql)
4950 .fetch_all(&pool)
4951 .await
4952 }
4953 }
4954 }
4955
4956 /// Like [`Self::raw`] but with BOUND parameters (gaps4 #24).
4957 ///
4958 /// `raw` sends its string verbatim and has no way to bind user input — the
4959 /// caller is on their own for escaping, which is an injection waiting to
4960 /// happen. `raw_with` binds each `values[i]` as a real parameter, so the
4961 /// placeholders in `sql` (`?` on SQLite, `$1..$N` on Postgres) carry data
4962 /// safely. Values are the framework's own `sea_query::Value`, so the same
4963 /// call is portable across both backends.
4964 ///
4965 /// ```ignore
4966 /// use umbral::db::query::Value; // sea_query::Value
4967 /// let rows: Vec<Report> = Report::objects()
4968 /// .raw_with(
4969 /// "SELECT * FROM report WHERE region = ? AND year >= ?",
4970 /// vec![Value::from(region), Value::from(year)],
4971 /// )
4972 /// .await?;
4973 /// ```
4974 ///
4975 /// Still a low-level escape hatch: it skips `select_related` /
4976 /// `prefetch_related`, and (like `raw`) routes to the READ pool — a raw
4977 /// statement that WRITES under a read/write split must pin the write pool
4978 /// with `.on(&pool)` / `.on_pg(&pool)`.
4979 pub async fn raw_with(
4980 &self,
4981 sql: &str,
4982 values: Vec<sea_query::Value>,
4983 ) -> Result<Vec<T>, sqlx::Error>
4984 where
4985 T: for<'r> sqlx::FromRow<'r, sqlx::sqlite::SqliteRow>
4986 + for<'r> sqlx::FromRow<'r, sqlx::postgres::PgRow>,
4987 {
4988 let bound = sea_query_binder::SqlxValues(sea_query::Values(values));
4989 let pool = resolve_pool::<T>(None, crate::db::RouteOp::Read);
4990 match pool {
4991 DbPool::Sqlite(pool) => {
4992 sqlx::query_as_with::<sqlx::Sqlite, T, _>(sql, bound)
4993 .fetch_all(&pool)
4994 .await
4995 }
4996 DbPool::Postgres(pool) => {
4997 sqlx::query_as_with::<sqlx::Postgres, T, _>(sql, bound)
4998 .fetch_all(&pool)
4999 .await
5000 }
5001 }
5002 }
5003
5004 /// `bulk_create` against an explicit Postgres pool.
5005 pub async fn bulk_create_pg(
5006 &self,
5007 instances: Vec<T>,
5008 pool: &sqlx::PgPool,
5009 ) -> Result<u64, crate::orm::write::WriteError>
5010 where
5011 T: serde::Serialize,
5012 {
5013 if instances.is_empty() {
5014 return Ok(0);
5015 }
5016 let maps: Result<Vec<_>, _> = instances.iter().map(serialize_to_map).collect();
5017 let maps = maps?;
5018 let stmt = build_insert_many_for::<T>("postgres", &maps)?;
5019 let (sql, values) = stmt.build_sqlx(PostgresQueryBuilder);
5020 let result = sqlx::query_with::<sqlx::Postgres, _>(&sql, values)
5021 .execute(pool)
5022 .await?;
5023 Ok(result.rows_affected())
5024 }
5025}
5026
5027// QuerySetTx (struct + impl) moved to `super::tx`; re-exported above.
5028
5029impl<T: Model> Manager<T> {
5030 /// Begin a new query on this manager attached to the given open transaction.
5031 ///
5032 /// Sugar for `T::objects().on_tx(tx)` — lets callers skip the intermediate
5033 /// `QuerySet` construction when they want to go straight to a terminal:
5034 ///
5035 /// ```rust,ignore
5036 /// umbral::db::transaction(|tx| async move {
5037 /// let post = Post::objects().on_tx(tx).create(new_post).await?;
5038 /// Ok::<_, MyError>(post)
5039 /// }).await?;
5040 /// ```
5041 pub fn on_tx<'a>(&self, tx: &'a mut crate::db::Transaction) -> QuerySetTx<'a, T> {
5042 self.queryset().on_tx(tx)
5043 }
5044
5045 /// INSERT one row inside `tx` and return the populated row.
5046 ///
5047 /// This is the primary Manager-level entry point for transactional writes.
5048 /// Equivalent to `Post::objects().on_tx(tx).create(instance)` but more
5049 /// ergonomic when you only need the one INSERT (no filter chain needed).
5050 ///
5051 /// ```rust,ignore
5052 /// umbral::db::transaction(|tx| async move {
5053 /// let post = Post::objects().create_in_tx(new_post, tx).await?;
5054 /// Ok::<_, MyError>(post)
5055 /// }).await?;
5056 /// ```
5057 pub async fn create_in_tx(
5058 &self,
5059 instance: T,
5060 tx: &mut crate::db::Transaction,
5061 ) -> Result<T, crate::orm::write::WriteError>
5062 where
5063 T: serde::Serialize
5064 + for<'r> sqlx::FromRow<'r, sqlx::sqlite::SqliteRow>
5065 + for<'r> sqlx::FromRow<'r, sqlx::postgres::PgRow>
5066 + HydrateRelated,
5067 {
5068 let map = serialize_to_map(&instance)?;
5069 let stmt = build_insert_one_for::<T>(tx.backend_name(), &map)?;
5070 match tx.backend_name() {
5071 "sqlite" => {
5072 let inner = tx.as_sqlite_mut().unwrap();
5073 let (sql, values) = stmt.build_sqlx(SqliteQueryBuilder);
5074 let mut row = sqlx::query_as_with::<sqlx::Sqlite, T, _>(&sql, values)
5075 .fetch_one(&mut **inner)
5076 .await?;
5077 row.set_m2m_parent_ids();
5078 Ok(row)
5079 }
5080 _ => {
5081 let inner = tx.as_pg_mut().unwrap();
5082 let (sql, values) = stmt.build_sqlx(PostgresQueryBuilder);
5083 let mut row = sqlx::query_as_with::<sqlx::Postgres, T, _>(&sql, values)
5084 .fetch_one(&mut **inner)
5085 .await?;
5086 row.set_m2m_parent_ids();
5087 Ok(row)
5088 }
5089 }
5090 }
5091
5092 /// INSERT many rows inside `tx`.
5093 ///
5094 /// Returns the number of rows inserted. Empty input is a no-op.
5095 pub async fn bulk_create_in_tx(
5096 &self,
5097 instances: Vec<T>,
5098 tx: &mut crate::db::Transaction,
5099 ) -> Result<u64, crate::orm::write::WriteError>
5100 where
5101 T: serde::Serialize,
5102 {
5103 if instances.is_empty() {
5104 return Ok(0);
5105 }
5106 let maps: Result<Vec<_>, _> = instances.iter().map(serialize_to_map).collect();
5107 let maps = maps?;
5108 let stmt = build_insert_many_for::<T>(tx.backend_name(), &maps)?;
5109 match tx.backend_name() {
5110 "sqlite" => {
5111 let inner = tx.as_sqlite_mut().unwrap();
5112 let (sql, values) = stmt.build_sqlx(SqliteQueryBuilder);
5113 let result = sqlx::query_with::<sqlx::Sqlite, _>(&sql, values)
5114 .execute(&mut **inner)
5115 .await?;
5116 Ok(result.rows_affected())
5117 }
5118 _ => {
5119 let inner = tx.as_pg_mut().unwrap();
5120 let (sql, values) = stmt.build_sqlx(PostgresQueryBuilder);
5121 let result = sqlx::query_with::<sqlx::Postgres, _>(&sql, values)
5122 .execute(&mut **inner)
5123 .await?;
5124 Ok(result.rows_affected())
5125 }
5126 }
5127 }
5128
5129 // =====================================================================
5130 // Per-instance signal-firing write methods.
5131 //
5132 // `save(instance)` and `delete_instance(instance)` are the methods that
5133 // fire the ORM lifecycle signals (`pre_save` / `post_save` /
5134 // `pre_delete` / `post_delete`). The existing bulk methods
5135 // (`create`, `bulk_create`, `QuerySet::update_values`,
5136 // `QuerySet::delete`) remain signal-free by design:
5137 // bulk operations bypass signals for performance.
5138 //
5139 // Signal name format: `<event>:<table>` — e.g. `post_save:post`.
5140 // Payload shapes:
5141 // save: `{ "instance": <M as JSON>, "created": bool }`
5142 // delete: `{ "instance": <M as JSON> }`
5143 //
5144 // The `created` flag on save follows the convention:
5145 // `true` when the PK is the default sentinel → INSERT path.
5146 // `false` when the PK is non-default → UPDATE path.
5147 // =====================================================================
5148
5149 /// Save one instance, firing `pre_save` + `post_save` signals.
5150 ///
5151 /// Determines INSERT vs UPDATE by checking whether the primary key
5152 /// is the autoincrement sentinel (`0` for integers, nil UUID, empty
5153 /// string). If it is, an INSERT is performed (`created = true`);
5154 /// otherwise an `UPDATE ... WHERE pk = <value>` is run (`created = false`).
5155 ///
5156 /// Returns the row as it exists in the database after the write
5157 /// (populated PK for inserts, same row for updates).
5158 ///
5159 /// ## Signal contract
5160 ///
5161 /// - `pre_save:<table>` fires before the database write with
5162 /// `{ "instance": ..., "created": bool, "actor": ... }`.
5163 /// - `post_save:<table>` fires after the database write with the
5164 /// DB-read-back row and the same envelope keys.
5165 ///
5166 /// The `"actor"` value is set by the nearest enclosing
5167 /// [`crate::signals::with_actor`] scope; `Value::Null` when no
5168 /// scope is active.
5169 ///
5170 /// ## Bulk paths fire bulk signals, not per-row signals
5171 ///
5172 /// `Manager::create`, `Manager::bulk_create`, and
5173 /// `QuerySet::update_values` / `QuerySet::delete` do NOT fire
5174 /// per-row `post_save` / `post_delete`. They fire
5175 /// `bulk_post_save:<table>` / `bulk_post_delete:<table>` once per
5176 /// statement with the affected PKs in the payload. Use `save` /
5177 /// `delete_instance` when per-row signal semantics are needed.
5178 pub async fn save(&self, instance: T) -> Result<T, crate::orm::write::SaveError>
5179 where
5180 T: serde::Serialize
5181 + for<'r> sqlx::FromRow<'r, sqlx::sqlite::SqliteRow>
5182 + for<'r> sqlx::FromRow<'r, sqlx::postgres::PgRow>,
5183 {
5184 use crate::orm::write::{SaveError, is_default_pk};
5185 // Determine INSERT vs UPDATE by inspecting the PK field.
5186 let pk_field = T::FIELDS
5187 .iter()
5188 .find(|f| f.primary_key)
5189 .ok_or(SaveError::NoPrimaryKey)?;
5190 let map = serialize_to_map(&instance).map_err(SaveError::Write)?;
5191 let pk_val = map
5192 .get(pk_field.name)
5193 .cloned()
5194 .unwrap_or(serde_json::Value::Null);
5195 let created = is_default_pk(pk_field.ty, &pk_val);
5196
5197 // Fire pre_save before the write.
5198 crate::signals::emit_pre_save::<T>(&instance, created).await;
5199
5200 let pool = resolve_pool::<T>(None, crate::db::RouteOp::Write);
5201 let backend = pool.backend_name();
5202
5203 if created {
5204 // INSERT path.
5205 let stmt = build_insert_one_for::<T>(backend, &map).map_err(SaveError::Write)?;
5206 let row = match pool {
5207 DbPool::Sqlite(pool) => {
5208 let (sql, values) = stmt.build_sqlx(SqliteQueryBuilder);
5209 sqlx::query_as_with::<sqlx::Sqlite, T, _>(&sql, values)
5210 .fetch_one(&pool)
5211 .await
5212 .map_err(|e| SaveError::Write(crate::orm::write::WriteError::Sqlx(e)))?
5213 }
5214 DbPool::Postgres(pool) => {
5215 let (sql, values) = stmt.build_sqlx(PostgresQueryBuilder);
5216 sqlx::query_as_with::<sqlx::Postgres, T, _>(&sql, values)
5217 .fetch_one(&pool)
5218 .await
5219 .map_err(|e| SaveError::Write(crate::orm::write::WriteError::Sqlx(e)))?
5220 }
5221 };
5222 // Fire post_save with the DB-populated row.
5223 crate::orm::audit::record(
5224 &crate::migrate::ModelMeta::for_::<T>(),
5225 &serde_json::to_value(&row)
5226 .ok()
5227 .and_then(|v| {
5228 v.as_object().map(|o| {
5229 crate::orm::audit::pk_of(&crate::migrate::ModelMeta::for_::<T>(), o)
5230 })
5231 })
5232 .unwrap_or_default(),
5233 crate::orm::audit::CREATE,
5234 None,
5235 serde_json::to_value(&row)
5236 .ok()
5237 .and_then(|v| v.as_object().cloned())
5238 .as_ref(),
5239 )
5240 .await;
5241 crate::signals::emit_post_save::<T>(&row, true).await;
5242 Ok(row)
5243 } else {
5244 // UPDATE path: UPDATE ... WHERE <pk> = <value> RETURNING *.
5245 use sea_query::{Alias, Expr, Query};
5246
5247 // gaps2 #92 — snapshot the pre-UPDATE row for `pre_update` /
5248 // `post_update` subscribers, but ONLY when one exists. The
5249 // extra SELECT-by-PK is gated on `has_subscribers` so the
5250 // common UPDATE path (no `*_update` listener) pays nothing.
5251 // Best-effort TOCTOU: the snapshot reads the row before the
5252 // UPDATE; a concurrent writer between the two is accepted.
5253 let pre_table = T::TABLE;
5254 let want_pre = crate::signals::has_subscribers(&format!("pre_update:{pre_table}"));
5255 let want_post = crate::signals::has_subscribers(&format!("post_update:{pre_table}"));
5256 let previous: Option<T> = if want_pre || want_post {
5257 let mut sel = Query::select();
5258 sel.from(crate::db::router::schema_qualified_table(T::TABLE));
5259 for field in T::FIELDS {
5260 sel.column(Alias::new(field.name));
5261 }
5262 let pk_sea_sel = crate::orm::write::json_to_sea_value(
5263 pk_field.ty,
5264 &pk_val,
5265 false,
5266 pk_field.name,
5267 None,
5268 )
5269 .map_err(SaveError::Write)?;
5270 sel.and_where(Expr::col(Alias::new(pk_field.name)).eq(pk_sea_sel));
5271 sel.limit(1);
5272 match pool {
5273 DbPool::Sqlite(ref pool) => {
5274 let (sql, values) = sel.build_sqlx(SqliteQueryBuilder);
5275 sqlx::query_as_with::<sqlx::Sqlite, T, _>(&sql, values)
5276 .fetch_optional(pool)
5277 .await
5278 .map_err(|e| SaveError::Write(crate::orm::write::WriteError::Sqlx(e)))?
5279 }
5280 DbPool::Postgres(ref pool) => {
5281 let (sql, values) = sel.build_sqlx(PostgresQueryBuilder);
5282 sqlx::query_as_with::<sqlx::Postgres, T, _>(&sql, values)
5283 .fetch_optional(pool)
5284 .await
5285 .map_err(|e| SaveError::Write(crate::orm::write::WriteError::Sqlx(e)))?
5286 }
5287 }
5288 } else {
5289 None
5290 };
5291 // Fire pre_update before the UPDATE when both a snapshot exists
5292 // and a subscriber wants it.
5293 if want_pre {
5294 if let Some(prev) = &previous {
5295 crate::signals::emit_pre_update::<T>(prev, &instance).await;
5296 }
5297 }
5298
5299 let mut stmt = Query::update();
5300 stmt.table(crate::db::router::schema_qualified_table(T::TABLE));
5301 for field in T::FIELDS {
5302 if field.primary_key {
5303 continue;
5304 }
5305 let val = map
5306 .get(field.name)
5307 .cloned()
5308 .unwrap_or(serde_json::Value::Null);
5309 let sea_val = crate::orm::write::json_to_sea_value(
5310 field.ty,
5311 &val,
5312 field.nullable,
5313 field.name,
5314 fk_pk_hint(field),
5315 )
5316 .map_err(SaveError::Write)?;
5317 stmt.value(Alias::new(field.name), sea_val);
5318 }
5319 // WHERE pk = <value>
5320 let pk_sea = crate::orm::write::json_to_sea_value(
5321 pk_field.ty,
5322 &pk_val,
5323 false,
5324 pk_field.name,
5325 None,
5326 )
5327 .map_err(SaveError::Write)?;
5328 stmt.and_where(Expr::col(Alias::new(pk_field.name)).eq(pk_sea));
5329 // RETURNING * so we can return the updated row.
5330 stmt.returning_all();
5331
5332 let row = match pool {
5333 DbPool::Sqlite(pool) => {
5334 let (sql, values) = stmt.build_sqlx(SqliteQueryBuilder);
5335 sqlx::query_as_with::<sqlx::Sqlite, T, _>(&sql, values)
5336 .fetch_one(&pool)
5337 .await
5338 .map_err(|e| SaveError::Write(crate::orm::write::WriteError::Sqlx(e)))?
5339 }
5340 DbPool::Postgres(pool) => {
5341 let (sql, values) = stmt.build_sqlx(PostgresQueryBuilder);
5342 sqlx::query_as_with::<sqlx::Postgres, T, _>(&sql, values)
5343 .fetch_one(&pool)
5344 .await
5345 .map_err(|e| SaveError::Write(crate::orm::write::WriteError::Sqlx(e)))?
5346 }
5347 };
5348 // Fire post_save with created=false.
5349 crate::signals::emit_post_save::<T>(&row, false).await;
5350 // gaps2 #92 — post_update carries the pre-UPDATE snapshot (old)
5351 // and the freshly-written row (new). Only when a subscriber
5352 // exists and the snapshot was captured.
5353 if want_post {
5354 if let Some(prev) = &previous {
5355 crate::signals::emit_post_update::<T>(prev, &row).await;
5356 }
5357 }
5358 Ok(row)
5359 }
5360 }
5361
5362 /// Delete one instance by primary key, firing `pre_delete` +
5363 /// `post_delete` signals.
5364 ///
5365 /// Issues `DELETE FROM <table> WHERE <pk> = <value>`. Returns the
5366 /// number of rows affected (0 if the row was already gone, 1 otherwise).
5367 ///
5368 /// ## Signal contract
5369 ///
5370 /// - `pre_delete:<table>` fires before the DELETE with
5371 /// `{ "instance": ..., "actor": ... }`.
5372 /// - `post_delete:<table>` fires after the DELETE with the same
5373 /// payload shape.
5374 ///
5375 /// The `"actor"` value is set by the nearest enclosing
5376 /// [`crate::signals::with_actor`] scope; `Value::Null` when no
5377 /// scope is active. The instance value passed to both signals is
5378 /// the value supplied by the caller — not a DB read-back. If you
5379 /// need the freshest DB state before deletion, fetch it first with
5380 /// `.get(...)` then pass to this method.
5381 ///
5382 /// ## Bulk paths fire bulk signals
5383 ///
5384 /// `QuerySet::delete()` (the filter-chain DELETE) fires
5385 /// `bulk_post_delete:<table>` with the list of affected PKs, not
5386 /// per-row `pre_delete` / `post_delete`. Use `delete_instance` for
5387 /// per-row signal semantics.
5388 pub async fn delete_instance(&self, instance: &T) -> Result<u64, crate::orm::write::SaveError>
5389 where
5390 T: serde::Serialize,
5391 {
5392 use crate::orm::write::SaveError;
5393 let pk_field = T::FIELDS
5394 .iter()
5395 .find(|f| f.primary_key)
5396 .ok_or(SaveError::NoPrimaryKey)?;
5397 let map = serialize_to_map(instance).map_err(SaveError::Write)?;
5398 let pk_val = map
5399 .get(pk_field.name)
5400 .cloned()
5401 .unwrap_or(serde_json::Value::Null);
5402
5403 // Fire pre_delete before the write.
5404 crate::signals::emit_pre_delete::<T>(instance).await;
5405
5406 let pk_sea =
5407 crate::orm::write::json_to_sea_value(pk_field.ty, &pk_val, false, pk_field.name, None)
5408 .map_err(SaveError::Write)?;
5409
5410 use sea_query::{Alias, Expr, Query};
5411 // Feature #72 — soft-delete redirect. For models tagged
5412 // `#[umbral(soft_delete)]`, set deleted_at instead of issuing
5413 // DELETE. Pre/post_delete signals still fire because the
5414 // logical contract ("this row is gone from the visible
5415 // table") is preserved — only the physical SQL changed.
5416 // Hard-delete is not exposed through delete_instance (it's
5417 // a typed per-row helper); call `QuerySet::filter(pk =
5418 // instance.id).hard_delete().delete()` when you need it.
5419 let stmt_sql = if T::SOFT_DELETE {
5420 let now = chrono::Utc::now();
5421 let mut up = Query::update();
5422 up.table(crate::db::router::schema_qualified_table(T::TABLE));
5423 up.value(
5424 Alias::new("deleted_at"),
5425 sea_query::Value::ChronoDateTimeUtc(Some(Box::new(now))),
5426 );
5427 up.and_where(Expr::col(Alias::new(pk_field.name)).eq(pk_sea));
5428 // Idempotency guard — don't bump an already-set timestamp.
5429 up.and_where(Expr::col(Alias::new("deleted_at")).is_null());
5430 SoftOrHardStatement::Update(up)
5431 } else {
5432 let mut stmt = Query::delete();
5433 stmt.from_table(crate::db::router::schema_qualified_table(T::TABLE));
5434 stmt.and_where(Expr::col(Alias::new(pk_field.name)).eq(pk_sea));
5435 SoftOrHardStatement::Delete(stmt)
5436 };
5437
5438 let pool = resolve_pool::<T>(None, crate::db::RouteOp::Write);
5439 let affected = match (&pool, stmt_sql) {
5440 (DbPool::Sqlite(pool), SoftOrHardStatement::Delete(stmt)) => {
5441 let (sql, values) = stmt.build_sqlx(SqliteQueryBuilder);
5442 sqlx::query_with::<sqlx::Sqlite, _>(&sql, values)
5443 .execute(pool)
5444 .await
5445 .map_err(|e| SaveError::Write(crate::orm::write::WriteError::Sqlx(e)))?
5446 .rows_affected()
5447 }
5448 (DbPool::Postgres(pool), SoftOrHardStatement::Delete(stmt)) => {
5449 let (sql, values) = stmt.build_sqlx(PostgresQueryBuilder);
5450 sqlx::query_with::<sqlx::Postgres, _>(&sql, values)
5451 .execute(pool)
5452 .await
5453 .map_err(|e| SaveError::Write(crate::orm::write::WriteError::Sqlx(e)))?
5454 .rows_affected()
5455 }
5456 (DbPool::Sqlite(pool), SoftOrHardStatement::Update(stmt)) => {
5457 let (sql, values) = stmt.build_sqlx(SqliteQueryBuilder);
5458 sqlx::query_with::<sqlx::Sqlite, _>(&sql, values)
5459 .execute(pool)
5460 .await
5461 .map_err(|e| SaveError::Write(crate::orm::write::WriteError::Sqlx(e)))?
5462 .rows_affected()
5463 }
5464 (DbPool::Postgres(pool), SoftOrHardStatement::Update(stmt)) => {
5465 let (sql, values) = stmt.build_sqlx(PostgresQueryBuilder);
5466 sqlx::query_with::<sqlx::Postgres, _>(&sql, values)
5467 .execute(pool)
5468 .await
5469 .map_err(|e| SaveError::Write(crate::orm::write::WriteError::Sqlx(e)))?
5470 .rows_affected()
5471 }
5472 };
5473
5474 // Fire post_delete after the write.
5475 crate::signals::emit_post_delete::<T>(instance).await;
5476
5477 Ok(affected)
5478 }
5479}
5480
5481/// Internal enum used by `Manager::delete_instance` to dispatch on
5482/// soft-delete vs hard-delete without duplicating the four backend ×
5483/// statement match arms inline. One variant per SQL shape.
5484enum SoftOrHardStatement {
5485 Delete(sea_query::DeleteStatement),
5486 Update(sea_query::UpdateStatement),
5487}
5488
5489// Hydration helpers (hydrate_select_related, hydrate_select_related_nested,
5490// hydrate_prefetch_related, hydrate_reverse_fk_for_field, fetch_related_as_json,
5491// fetch_reverse_fk_children) moved to `super::hydration`.
5492
5493// Insert builders (serialize_to_map, build_insert_one_for, build_insert_many_for)
5494// and pk_field moved to `super::write_helpers`.
5495
5496// `decode_agg_sqlite` / `decode_agg_pg` moved to
5497// `backend_sqlite::decode_agg` / `backend_pg::decode_agg`.
5498
5499// =========================================================================
5500// #113: M2M-via-JOIN dedup decoders
5501//
5502// When .join_related() includes one or more M2M fields, the result
5503// set has one row per (parent, child) combo, so a parent with N
5504// matching children appears N times. The caller wants ONE T per
5505// parent with the M2M slot populated.
5506//
5507// The algorithm:
5508// - First time we see a parent PK: decode T via FromRow, hydrate
5509// any FK joins from this row.
5510// - Every subsequent row for the same parent: extract the M2M
5511// child JsonValue (or skip on LEFT JOIN miss).
5512// - Dedup children by (parent_pk, field, child_pk) so that
5513// joining TWO M2Ms doesn't multiply the child sets (the JOIN
5514// produces parent × m2m1 × m2m2 rows).
5515// - Hand each parent its M2M buckets via set_m2m_resolved_json.
5516// =========================================================================
5517
5518fn dedup_decode_sqlite<T: Model + HydrateRelated>(
5519 raw_rows: &[sqlx::sqlite::SqliteRow],
5520 fk_join_fields: &[String],
5521 m2m_join_fields: &[String],
5522) -> Result<Vec<T>, sqlx::Error>
5523where
5524 T: for<'r> sqlx::FromRow<'r, sqlx::sqlite::SqliteRow>,
5525{
5526 // PK-agnostic dedup: read the parent PK back through the shape-aware
5527 // decoder (using the parent model's PK SqlType) and key by `pk_key`,
5528 // so i64 / String / Uuid parents all dedup correctly.
5529 let parent_pk_col = crate::migrate::ModelMeta::for_::<T>()
5530 .fields
5531 .into_iter()
5532 .find(|c| c.primary_key)
5533 .ok_or_else(|| {
5534 sqlx::Error::Protocol(format!(
5535 "umbral::orm::join_related: model `{}` has no primary key, M2M JOIN \
5536 dedup requires one",
5537 T::NAME
5538 ))
5539 })?;
5540 let registered = crate::migrate::registered_models();
5541 let mut typed: Vec<T> = Vec::new();
5542 let mut idx_by_pk: HashMap<String, usize> = HashMap::new();
5543 // (parent_pk_key, field) → Vec<JsonValue> + a Set of seen child PK keys.
5544 let mut buckets: HashMap<(String, String), Vec<JsonValue>> = HashMap::new();
5545 let mut seen_children: HashMap<(String, String), std::collections::HashSet<String>> =
5546 HashMap::new();
5547 for row in raw_rows {
5548 let Ok(parent_json) = crate::orm::dynamic::decode_to_json(row, &parent_pk_col) else {
5549 continue;
5550 };
5551 let parent_key = crate::orm::pk_key(&parent_json);
5552 if let std::collections::hash_map::Entry::Vacant(e) = idx_by_pk.entry(parent_key.clone()) {
5553 let mut t = <T as sqlx::FromRow<_>>::from_row(row)?;
5554 backend_sqlite::hydrate_joined_rels::<T>(&mut t, row, fk_join_fields)?;
5555 e.insert(typed.len());
5556 typed.push(t);
5557 }
5558 for m2m_field in m2m_join_fields {
5559 // The M2M slot is keyed by the FIRST segment (the M2M field
5560 // name); the full `m2m_field` path may carry an onward FK
5561 // chain (`"tags__category"`) the child decoder nests.
5562 let m2m_seg = m2m_field.split("__").next().unwrap_or(m2m_field.as_str());
5563 let Some(rel) = T::M2M_RELATIONS.iter().find(|r| r.field_name == m2m_seg) else {
5564 continue;
5565 };
5566 let Some(child_meta) = registered.iter().find(|m| m.table == rel.target_table) else {
5567 continue;
5568 };
5569 let Some(child_json) =
5570 backend_sqlite::extract_m2m_child_json::<T>(row, m2m_field, child_meta)?
5571 else {
5572 continue;
5573 };
5574 // Dedup by child PK (PK-agnostic) so multi-M2M cartesian
5575 // doesn't duplicate this field's children.
5576 let child_key = child_json
5577 .as_object()
5578 .and_then(|m| {
5579 let pk_col = child_meta.fields.iter().find(|c| c.primary_key)?;
5580 m.get(&pk_col.name).map(crate::orm::pk_key)
5581 })
5582 .unwrap_or_default();
5583 let key = (parent_key.clone(), m2m_seg.to_string());
5584 let seen = seen_children.entry(key.clone()).or_default();
5585 if seen.insert(child_key) {
5586 buckets.entry(key).or_default().push(child_json);
5587 }
5588 }
5589 }
5590 for ((parent_key, field), children) in buckets {
5591 if let Some(&idx) = idx_by_pk.get(&parent_key) {
5592 typed[idx].set_m2m_resolved_json(&field, children);
5593 }
5594 }
5595 // LEFT JOIN miss handling: walk every (parent, field) pair
5596 // we expected to populate and zero-init any slot that never
5597 // got a hit. Without this a parent with no matching M2M
5598 // children would leave its slot None — distinguishable from
5599 // "loaded, empty" only by callers checking the absence.
5600 for (parent_key, &idx) in idx_by_pk.iter() {
5601 for field in m2m_join_fields {
5602 let seg = field.split("__").next().unwrap_or(field.as_str());
5603 if !seen_children.contains_key(&(parent_key.clone(), seg.to_string())) {
5604 typed[idx].set_m2m_resolved_json(seg, Vec::new());
5605 }
5606 }
5607 }
5608 Ok(typed)
5609}
5610
5611fn dedup_decode_pg<T: Model + HydrateRelated>(
5612 raw_rows: &[sqlx::postgres::PgRow],
5613 fk_join_fields: &[String],
5614 m2m_join_fields: &[String],
5615) -> Result<Vec<T>, sqlx::Error>
5616where
5617 T: for<'r> sqlx::FromRow<'r, sqlx::postgres::PgRow>,
5618{
5619 // PK-agnostic dedup — see the SQLite variant.
5620 let parent_pk_col = crate::migrate::ModelMeta::for_::<T>()
5621 .fields
5622 .into_iter()
5623 .find(|c| c.primary_key)
5624 .ok_or_else(|| {
5625 sqlx::Error::Protocol(format!(
5626 "umbral::orm::join_related: model `{}` has no primary key, M2M JOIN \
5627 dedup requires one",
5628 T::NAME
5629 ))
5630 })?;
5631 let registered = crate::migrate::registered_models();
5632 let mut typed: Vec<T> = Vec::new();
5633 let mut idx_by_pk: HashMap<String, usize> = HashMap::new();
5634 let mut buckets: HashMap<(String, String), Vec<JsonValue>> = HashMap::new();
5635 let mut seen_children: HashMap<(String, String), std::collections::HashSet<String>> =
5636 HashMap::new();
5637 for row in raw_rows {
5638 let Ok(parent_json) = crate::orm::dynamic::decode_pg_to_json(row, &parent_pk_col) else {
5639 continue;
5640 };
5641 let parent_key = crate::orm::pk_key(&parent_json);
5642 if let std::collections::hash_map::Entry::Vacant(e) = idx_by_pk.entry(parent_key.clone()) {
5643 let mut t = <T as sqlx::FromRow<_>>::from_row(row)?;
5644 backend_pg::hydrate_joined_rels::<T>(&mut t, row, fk_join_fields)?;
5645 e.insert(typed.len());
5646 typed.push(t);
5647 }
5648 for m2m_field in m2m_join_fields {
5649 let m2m_seg = m2m_field.split("__").next().unwrap_or(m2m_field.as_str());
5650 let Some(rel) = T::M2M_RELATIONS.iter().find(|r| r.field_name == m2m_seg) else {
5651 continue;
5652 };
5653 let Some(child_meta) = registered.iter().find(|m| m.table == rel.target_table) else {
5654 continue;
5655 };
5656 let Some(child_json) =
5657 backend_pg::extract_m2m_child_json::<T>(row, m2m_field, child_meta)?
5658 else {
5659 continue;
5660 };
5661 let child_key = child_json
5662 .as_object()
5663 .and_then(|m| {
5664 let pk_col = child_meta.fields.iter().find(|c| c.primary_key)?;
5665 m.get(&pk_col.name).map(crate::orm::pk_key)
5666 })
5667 .unwrap_or_default();
5668 let key = (parent_key.clone(), m2m_seg.to_string());
5669 let seen = seen_children.entry(key.clone()).or_default();
5670 if seen.insert(child_key) {
5671 buckets.entry(key).or_default().push(child_json);
5672 }
5673 }
5674 }
5675 for ((parent_key, field), children) in buckets {
5676 if let Some(&idx) = idx_by_pk.get(&parent_key) {
5677 typed[idx].set_m2m_resolved_json(&field, children);
5678 }
5679 }
5680 // Same LEFT JOIN miss zero-init as the SQLite path.
5681 for (parent_key, &idx) in idx_by_pk.iter() {
5682 for field in m2m_join_fields {
5683 let seg = field.split("__").next().unwrap_or(field.as_str());
5684 if !seen_children.contains_key(&(parent_key.clone(), seg.to_string())) {
5685 typed[idx].set_m2m_resolved_json(seg, Vec::new());
5686 }
5687 }
5688 }
5689 Ok(typed)
5690}