Skip to main content

umbral_core/orm/queryset/
mod.rs

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