Skip to main content

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