fraiseql_db/postgres/adapter/
relay.rs

1//! `RelayDatabaseAdapter` implementation for `PostgresAdapter`.
2
3use fraiseql_error::{FraiseQLError, Result};
4
5use super::{PostgresAdapter, escape_jsonb_key};
6use crate::{
7    dialect::PostgresDialect,
8    identifier::quote_postgres_identifier,
9    postgres::where_generator::PostgresWhereGenerator,
10    traits::{CursorValue, RelayDatabaseAdapter, RelayPageResult},
11    types::{
12        QueryParam,
13        sql_hints::{OrderByClause, OrderDirection},
14    },
15    where_clause::WhereClause,
16};
17
18impl RelayDatabaseAdapter for PostgresAdapter {
19    /// Execute keyset (cursor-based) pagination against a JSONB view.
20    ///
21    /// # `totalCount` semantics
22    ///
23    /// When `include_total_count` is `true`, **two queries** are issued on the same
24    /// connection:
25    ///
26    /// 1. A count query — `SELECT COUNT(*) FROM {view} WHERE {user_filter}` — that reflects the
27    ///    **full connection** size, ignoring cursor position. This is required by the Relay Cursor
28    ///    Connections spec, which defines `totalCount` as the count of all objects in the
29    ///    connection, regardless of `after`/`before`.
30    ///
31    /// 2. A page query — the cursor-filtered, limited result set.
32    ///
33    /// The two-query approach fixes a previous bug where `COUNT(*) OVER()` ran
34    /// inside the cursor-filtered subquery, causing `totalCount` to shrink as the
35    /// cursor advanced.  It also handles the edge case where the current page is
36    /// empty but the total count is non-zero (e.g., cursor past the last row).
37    ///
38    /// When `include_total_count` is `false`, only the page query is issued.
39    ///
40    /// # Performance note
41    ///
42    /// The count query scans all rows matching the user filter without LIMIT. On
43    /// large unfiltered tables this may be slow. Mitigations:
44    /// - Only enable `totalCount` when the client explicitly requests it (enforced by the executor
45    ///   via `include_total_count`).
46    /// - Add a `statement_timeout` on the connection for relay queries on very large datasets.
47    /// - Maintain a denormalised count table or materialised view for hot paths.
48    async fn execute_relay_page(
49        &self,
50        view: &str,
51        cursor_column: &str,
52        after: Option<CursorValue>,
53        before: Option<CursorValue>,
54        limit: u32,
55        forward: bool,
56        where_clause: Option<&WhereClause>,
57        order_by: Option<&[OrderByClause]>,
58        include_total_count: bool,
59    ) -> Result<RelayPageResult> {
60        let client = self.acquire_connection_with_retry().await?;
61        self.run_relay_page(
62            &**client,
63            view,
64            cursor_column,
65            after,
66            before,
67            limit,
68            forward,
69            where_clause,
70            order_by,
71            include_total_count,
72        )
73        .await
74    }
75
76    #[allow(clippy::too_many_arguments)] // Reason: relay pagination requires all cursor/filter/sort/count arguments plus session vars; no natural grouping
77    async fn execute_relay_page_with_session(
78        &self,
79        view: &str,
80        cursor_column: &str,
81        after: Option<CursorValue>,
82        before: Option<CursorValue>,
83        limit: u32,
84        forward: bool,
85        where_clause: Option<&WhereClause>,
86        order_by: Option<&[OrderByClause]>,
87        include_total_count: bool,
88        session_vars: &[(&str, &str)],
89    ) -> Result<RelayPageResult> {
90        // Fast path: no session vars => identical to execute_relay_page.
91        if session_vars.is_empty() {
92            return self
93                .execute_relay_page(
94                    view,
95                    cursor_column,
96                    after,
97                    before,
98                    limit,
99                    forward,
100                    where_clause,
101                    order_by,
102                    include_total_count,
103                )
104                .await;
105        }
106
107        // Apply set_config and run BOTH the page and count queries inside one
108        // transaction on one connection so RLS sees the session variables.
109        let mut client = self.acquire_connection_with_retry().await?;
110        let txn =
111            client.build_transaction().start().await.map_err(|e| FraiseQLError::Database {
112                message:   format!("Failed to start relay session-var transaction: {e}"),
113                sql_state: e.code().map(|c| c.code().to_string()),
114            })?;
115        super::database::apply_session_vars(&txn, session_vars).await?;
116        let result = self
117            .run_relay_page(
118                &*txn,
119                view,
120                cursor_column,
121                after,
122                before,
123                limit,
124                forward,
125                where_clause,
126                order_by,
127                include_total_count,
128            )
129            .await?;
130        txn.commit().await.map_err(|e| FraiseQLError::Database {
131            message:   format!("Failed to commit relay session-var transaction: {e}"),
132            sql_state: e.code().map(|c| c.code().to_string()),
133        })?;
134        Ok(result)
135    }
136}
137
138impl PostgresAdapter {
139    /// Build and run the relay page (and optional total-count) queries against
140    /// an arbitrary client — a pooled connection for the plain path, or a
141    /// transaction for the connection-affine `*_with_session` path.
142    #[allow(clippy::too_many_arguments)] // Reason: relay pagination requires all cursor/filter/sort/count arguments; no natural grouping
143    async fn run_relay_page<C>(
144        &self,
145        client: &C,
146        view: &str,
147        cursor_column: &str,
148        after: Option<CursorValue>,
149        before: Option<CursorValue>,
150        limit: u32,
151        forward: bool,
152        where_clause: Option<&WhereClause>,
153        order_by: Option<&[OrderByClause]>,
154        include_total_count: bool,
155    ) -> Result<RelayPageResult>
156    where
157        C: tokio_postgres::GenericClient + Sync,
158    {
159        let quoted_view = quote_postgres_identifier(view);
160        let quoted_col = quote_postgres_identifier(cursor_column);
161
162        // ── Cursor condition (page query only, NOT the count query) ────────────
163        //
164        // Per the Relay spec, totalCount ignores cursor position. The cursor
165        // condition is therefore excluded from the count query.
166        //
167        // The cursor occupies at most one parameter slot ($1) at the front of the
168        // page query's parameter list.
169        //
170        // UUID cursors use `$1::uuid` cast; BIGINT cursors use plain `$1`.
171        let cursor_param: Option<QueryParam>;
172        let cursor_where_part: Option<String>;
173        let active_cursor = if forward { after } else { before };
174        match active_cursor {
175            None => {
176                cursor_param = None;
177                cursor_where_part = None;
178            },
179            Some(CursorValue::Int64(pk)) => {
180                let op = if forward { ">" } else { "<" };
181                cursor_param = Some(QueryParam::BigInt(pk));
182                cursor_where_part = Some(format!("{quoted_col} {op} $1"));
183            },
184            Some(CursorValue::Uuid(uuid)) => {
185                let op = if forward { ">" } else { "<" };
186                cursor_param = Some(QueryParam::Text(uuid));
187                cursor_where_part = Some(format!("{quoted_col} {op} $1::uuid"));
188            },
189        }
190        let cursor_param_count: usize = usize::from(cursor_param.is_some());
191
192        // ── User WHERE clause ──────────────────────────────────────────────────
193        //
194        // Used in BOTH the count query (offset 0) and the page query (offset by
195        // cursor_param_count so parameter indices don't collide).
196        let mut user_where_json_params: Vec<serde_json::Value> = Vec::new();
197        let page_user_where_sql: Option<String> = if let Some(clause) = where_clause {
198            let generator = PostgresWhereGenerator::new(PostgresDialect);
199            let (sql, params) = generator.generate_with_param_offset(clause, cursor_param_count)?;
200            user_where_json_params = params;
201            Some(sql)
202        } else {
203            None
204        };
205        let user_param_count = user_where_json_params.len();
206
207        // ── ORDER BY clause ────────────────────────────────────────────────────
208        //
209        // Custom sort columns first, then cursor column as tiebreaker for stable
210        // keyset pagination.
211        let order_sql = if let Some(clauses) = order_by {
212            let mut parts: Vec<String> = clauses
213                .iter()
214                .map(|c| {
215                    let dir = match c.direction {
216                        OrderDirection::Asc => "ASC",
217                        OrderDirection::Desc => "DESC",
218                    };
219                    // escape_jsonb_key is defense-in-depth: field names are already
220                    // validated as GraphQL identifiers (which cannot contain `'`).
221                    format!("data->>'{field}' {dir}", field = escape_jsonb_key(&c.field))
222                })
223                .collect();
224            let primary_dir = if forward { "ASC" } else { "DESC" };
225            parts.push(format!("{quoted_col} {primary_dir}"));
226            format!(" ORDER BY {}", parts.join(", "))
227        } else {
228            let dir = if forward { "ASC" } else { "DESC" };
229            format!(" ORDER BY {quoted_col} {dir}")
230        };
231
232        // ── Page WHERE SQL ─────────────────────────────────────────────────────
233        //
234        // Combines cursor condition AND user filter with offset parameter indices.
235        let cursor_part = cursor_where_part.as_deref().unwrap_or("");
236        let user_part =
237            page_user_where_sql.as_deref().map(|s| format!("({s})")).unwrap_or_default();
238        let page_where_sql = if cursor_part.is_empty() && user_part.is_empty() {
239            String::new()
240        } else if cursor_part.is_empty() {
241            format!(" WHERE {user_part}")
242        } else if user_part.is_empty() {
243            format!(" WHERE {cursor_part}")
244        } else {
245            format!(" WHERE {cursor_part} AND {user_part}")
246        };
247
248        // ── LIMIT parameter index ──────────────────────────────────────────────
249        let limit_idx = cursor_param_count + user_param_count + 1;
250
251        // ── Page SQL ───────────────────────────────────────────────────────────
252        //
253        // Backward pagination wraps the inner query in a subquery to re-sort
254        // the descending page back to ascending order.
255        let page_sql = if forward {
256            format!("SELECT data FROM {quoted_view}{page_where_sql}{order_sql} LIMIT ${limit_idx}")
257        } else {
258            let inner = format!(
259                "SELECT data, {quoted_col} AS _relay_cursor \
260                 FROM {quoted_view}{page_where_sql}{order_sql} LIMIT ${limit_idx}"
261            );
262            format!("SELECT data FROM ({inner}) _relay_page ORDER BY _relay_cursor ASC")
263        };
264
265        // ── Page params: [cursor?, user_where_params..., limit] ────────────────
266        let mut page_typed_params: Vec<QueryParam> = Vec::new();
267        if let Some(cp) = cursor_param {
268            page_typed_params.push(cp);
269        }
270        for v in &user_where_json_params {
271            page_typed_params.push(QueryParam::from(v.clone()));
272        }
273        page_typed_params.push(QueryParam::BigInt(i64::from(limit)));
274
275        // ── Execute page query (on the caller-provided client / transaction) ────
276        let page_param_refs = crate::types::as_sql_param_refs(&page_typed_params);
277
278        let page_rows = client.query(&page_sql, &page_param_refs).await.map_err(|e| {
279            FraiseQLError::Database {
280                message:   e.to_string(),
281                sql_state: e.code().map(|c| c.code().to_string()),
282            }
283        })?;
284
285        let rows: Vec<crate::types::JsonbValue> = page_rows
286            .iter()
287            .map(|row| super::jsonb_cell(row, "data", &page_sql))
288            .collect::<Result<Vec<_>>>()?;
289
290        // ── Count query (Relay spec: totalCount ignores cursor position) ────────
291        //
292        // The WHERE clause is regenerated with offset 0 (no cursor parameter prefix)
293        // because this is a standalone query. Using the same connection avoids an
294        // extra pool acquisition.
295        let total_count = if include_total_count {
296            let (count_sql, count_typed_params) = if let Some(clause) = where_clause {
297                let generator = PostgresWhereGenerator::new(PostgresDialect);
298                let (where_sql, params) = generator.generate_with_param_offset(clause, 0)?;
299                let sql = format!("SELECT COUNT(*) FROM {quoted_view} WHERE ({where_sql})");
300                let typed: Vec<QueryParam> = params.into_iter().map(QueryParam::from).collect();
301                (sql, typed)
302            } else {
303                (format!("SELECT COUNT(*) FROM {quoted_view}"), Vec::<QueryParam>::new())
304            };
305
306            let count_param_refs = crate::types::as_sql_param_refs(&count_typed_params);
307
308            let count_row = client.query_one(&count_sql, &count_param_refs).await.map_err(|e| {
309                FraiseQLError::Database {
310                    message:   e.to_string(),
311                    sql_state: e.code().map(|c| c.code().to_string()),
312                }
313            })?;
314
315            let total: i64 = count_row.get(0);
316            // cast_unsigned() is the clippy-recommended alternative to `as u64` for i64;
317            // it has the same bit-pattern semantics but makes the sign-loss intent explicit.
318            // Row counts from COUNT(*) are always non-negative so sign loss is impossible.
319            Some(total.cast_unsigned())
320        } else {
321            None
322        };
323
324        Ok(RelayPageResult::new(rows, total_count))
325    }
326}
fraiseql_db/postgres/adapter/relay.rs

fraiseql_db/postgres/adapter/
relay.rs
