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