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