postrust_server/app.rs
1//! Request handling.
2
3use crate::state::AppState;
4use axum::{
5 body::Body,
6 extract::{Request, State},
7 http::StatusCode,
8 response::{IntoResponse, Response},
9};
10use postrust_auth::authenticate;
11use postrust_core::{
12 create_action_plan, parse_request, ActionPlan, ApiRequest, CallPlan, DbActionPlan,
13};
14use postrust_response::{format_response, QueryResult, Response as PgrstResponse};
15use std::sync::Arc;
16use tracing::{debug, error};
17
18/// Column carrying the parent's whole row, for computed relationships.
19///
20/// A computed relationship is a function taking the parent row. By the time an
21/// embed expression runs, the parent is a derived table, and a derived table's
22/// alias has type `record` -- which PostgreSQL will not cast to the table's
23/// composite type. So the row is captured one level down, where the real table
24/// is still in scope and a bare reference to it yields the composite, and the
25/// embed expression reads it from there. It is stripped from the response like
26/// any other column added for embedding.
27///
28/// Defined in `postrust-core` beside the `RETURNING` that emits it for a
29/// mutation, so the two spellings cannot drift.
30use postrust_core::query::PARENT_ROW_COLUMN;
31
32/// Marks an embedded object whose columns belong to its parent.
33///
34/// A spread has no key of its own in the response -- its columns land in the
35/// parent object. In SQL it is still one JSON column like any other embed, so
36/// it is given a name nothing can collide with and dissolved into its parent
37/// once the rows are JSON. Doing it there rather than in SQL is what lets a
38/// spread nest inside another, and lets it spread a computed relationship.
39const SPREAD_KEY_PREFIX: &str = "pgrst_spread_";
40const PARENT_ROW_COLUMN_REF: &str = "\"src\".\"pgrst_parent_row\"";
41
42/// Main request handler.
43pub async fn handle_request(State(state): State<Arc<AppState>>, request: Request) -> Response {
44 let method = request.method().clone();
45 let path = request.uri().path().to_string();
46
47 debug!("{} {}", method, path);
48
49 let verbatim_db_errors = state.config.compat_mode;
50
51 match process_request(state, request).await {
52 Ok(response) => response.into_response(),
53 Err(e) => error_response(e, verbatim_db_errors).into_response(),
54 }
55}
56
57/// Process a request and return a response.
58async fn process_request(
59 state: Arc<AppState>,
60 request: Request,
61) -> Result<Response, postrust_core::Error> {
62 // Extract auth header
63 let auth_header = request
64 .headers()
65 .get("authorization")
66 .and_then(|v| v.to_str().ok());
67
68 // Authenticate
69 let auth_result = authenticate(auth_header, &state.jwt_config).map_err(jwt_error)?;
70
71 debug!("Authenticated as role: {}", auth_result.role);
72
73 // Parse request
74 let (parts, body) = request.into_parts();
75 let body_bytes = axum::body::to_bytes(body, 10 * 1024 * 1024)
76 .await
77 .map_err(|e| postrust_core::Error::InvalidBody(e.to_string()))?;
78
79 // Build HTTP request for parsing
80 let mut builder = http::Request::builder()
81 .method(parts.method.clone())
82 .uri(parts.uri.clone());
83
84 for (key, value) in &parts.headers {
85 builder = builder.header(key, value);
86 }
87
88 let http_request = builder
89 .body(body_bytes.clone())
90 .map_err(|e| postrust_core::Error::Internal(e.to_string()))?;
91
92 // Parse API request
93 let mut api_request = parse_request(
94 &http_request,
95 state.default_schema(),
96 state.schemas(),
97 state.config.db_max_rows,
98 )?;
99
100 // Parse payload
101 if !body_bytes.is_empty() {
102 let payload = postrust_core::api_request::payload::parse_payload(
103 body_bytes,
104 &api_request.content_media_type,
105 )?;
106 api_request.payload = payload;
107 }
108
109 // A body that names different columns row by row cannot be written by one
110 // statement. Skipped when `?columns=` says which columns to write, since
111 // then the rows are free to differ.
112 if api_request.query_params.columns.is_none() {
113 if let Some(payload) = &api_request.payload {
114 postrust_core::api_request::payload::validate_uniform_keys(payload)?;
115 }
116 }
117
118 // A write with nothing to write is refused here rather than left to
119 // produce an `UPDATE ... SET` with no assignments, which PostgreSQL
120 // reports as a syntax error -- true, and no use at all to the client.
121 if api_request.payload.is_none() && needs_payload(&api_request) {
122 return Err(postrust_core::Error::InvalidBody(
123 "Empty or invalid json".into(),
124 ));
125 }
126
127 // `application/vnd.pgrst.plan` asks for the query plan instead of the
128 // result. It is off unless a server turns it on -- it says a great deal
129 // about the schema and the data -- and a server that has not is required
130 // to say so rather than answer with something else.
131 //
132 // Checked after the body, not before it: a request that could not be read
133 // at all has not got as far as asking for anything, and PostgREST answers
134 // it with what is wrong with the request rather than with what the server
135 // declines to serve.
136 if let Some(plan) = requested_plan(&api_request) {
137 return Err(postrust_core::Error::InvalidMediaType(plan));
138 }
139
140 // Get schema cache
141 let schema_cache = state.schema_cache().await;
142
143 // Embedding joins on a column of the parent row, so that column has to be
144 // selected even when the client did not ask for it. Any column added this
145 // way is removed from the response once the embed is attached.
146 let added_join_columns = add_embed_join_columns(&mut api_request, &schema_cache)?;
147
148 if !state.config.db_aggregates_enabled && selects_an_aggregate(&api_request) {
149 return Err(postrust_core::Error::AggregatesNotAllowed);
150 }
151
152 // Create execution plan
153 let plan = create_action_plan(&api_request, &schema_cache)?;
154
155 // Execute plan
156 let result = execute_plan(
157 &state,
158 &api_request,
159 &plan,
160 &auth_result,
161 &added_join_columns,
162 )
163 .await?;
164
165 // Format response
166 // A singular response the result cannot satisfy is the client's business,
167 // not an internal failure: it asked for one object and the query answered
168 // with some other number of rows.
169 let response = format_response(&api_request, &result).map_err(|e| match e {
170 postrust_response::FormatError::MultipleRows | postrust_response::FormatError::NotFound => {
171 // How many rows there were, not how many were kept: a write the
172 // caller wanted no representation from still touched them, and
173 // that count is what the client got wrong.
174 postrust_core::Error::NotSingular {
175 rows: result.rows.len().max(result.affected),
176 }
177 }
178 other => postrust_core::Error::Internal(other.to_string()),
179 })?;
180
181 Ok(build_response(response))
182}
183
184/// The plan media type a request asked for, when that is all it will accept.
185///
186/// A request that would also take JSON gets JSON; one that will take only the
187/// plan is refused, and the message names what it asked for.
188fn requested_plan(api_request: &ApiRequest) -> Option<String> {
189 use postrust_core::api_request::MediaType;
190
191 // Named back with their parameters, since that is what was asked for: a
192 // plan for CSV and a plan for JSON are different requests, and a client
193 // told only that "application/vnd.pgrst.plan+json" is unavailable has not
194 // been told which of its requests was refused.
195 let accepted: Vec<String> = api_request
196 .accept_media_types
197 .iter()
198 .filter(|media| matches!(media, MediaType::Plan { .. }))
199 .map(MediaType::to_mime)
200 .collect();
201
202 match accepted.len() == api_request.accept_media_types.len() && !accepted.is_empty() {
203 true => Some(accepted.join(", ")),
204 false => None,
205 }
206}
207
208/// Which kind of mutation this request is, if it is one.
209fn mutation_kind(api_request: &ApiRequest) -> Option<postrust_core::api_request::Mutation> {
210 use postrust_core::api_request::{Action, DbAction};
211
212 match &api_request.action {
213 Action::Db(DbAction::RelationMut { mutation, .. }) => Some(mutation.clone()),
214 _ => None,
215 }
216}
217
218/// Split a leading `WITH ... AS (...)` off the front of a statement.
219///
220/// Returns the clause -- with its trailing space, or empty when there is none
221/// -- and the statement that follows it. A `WITH` containing an `INSERT`,
222/// `UPDATE` or `DELETE` may only appear at the top level, so anything that
223/// wraps the statement has to wrap what follows the clause and put the clause
224/// back in front.
225fn split_leading_cte(sql: &str) -> (String, &str) {
226 let Some(open) = sql.strip_prefix("WITH ").and_then(|rest| rest.find('(')) else {
227 return (String::new(), sql);
228 };
229 let open = open + "WITH ".len();
230
231 let mut depth = 0usize;
232 let mut quoted = None::<char>;
233 for (index, ch) in sql[open..].char_indices() {
234 match (quoted, ch) {
235 (Some(quote), ch) if ch == quote => quoted = None,
236 (Some(_), _) => {}
237 (None, '\'') | (None, '"') => quoted = Some(ch),
238 (None, '(') => depth += 1,
239 (None, ')') => {
240 depth -= 1;
241 if depth == 0 {
242 let end = open + index + 1;
243 return (format!("{} ", &sql[..end]), sql[end..].trim_start());
244 }
245 }
246 _ => {}
247 }
248 }
249
250 (String::new(), sql)
251}
252
253/// Render a map of strings as a JSON object.
254///
255/// PostgreSQL will not accept every header name as a setting name -- a `-` is
256/// not allowed in one from version 14 -- so the whole map goes into a single
257/// setting as JSON and a function picks out what it wants with `->>`.
258fn json_object<'a>(entries: impl IntoIterator<Item = (&'a String, &'a String)>) -> String {
259 serde_json::Value::Object(
260 entries
261 .into_iter()
262 .map(|(name, value)| (name.clone(), serde_json::Value::String(value.clone())))
263 .collect(),
264 )
265 .to_string()
266}
267
268/// The prefix given to primary key columns added only to build `Location`.
269const LOCATION_KEY_PREFIX: &str = "pgrst_location_";
270
271/// The column a media handler's statement carries its row count in.
272///
273/// Named like the other bookkeeping columns so it cannot collide with anything
274/// the request selected. Never reaches the client: the response body is the
275/// rendered value, read from the first column.
276const MEDIA_ROW_COUNT_COLUMN: &str = "pgrst_media_rows";
277
278/// Whether this request writes the row its URL names.
279fn is_upsert(api_request: &ApiRequest) -> bool {
280 use postrust_core::api_request::{Action, DbAction, Mutation};
281
282 matches!(
283 &api_request.action,
284 Action::Db(DbAction::RelationMut {
285 mutation: Mutation::SingleUpsert,
286 ..
287 })
288 )
289}
290
291/// The `Location` of a newly created row, and removal of the key columns that
292/// were added to find it.
293///
294/// `/projects?id=eq.7` -- the address the row can be read back from, which is
295/// what a 201 is required to give. A key column that came back null is
296/// addressed with `is.null`, since `eq.` would match nothing.
297fn build_location(
298 api_request: &ApiRequest,
299 rows: &mut [serde_json::Value],
300 keys: &[String],
301) -> Option<String> {
302 if keys.is_empty() {
303 return None;
304 }
305
306 let table = api_request.path.rsplit('/').next()?;
307 let mut conditions = Vec::with_capacity(keys.len());
308
309 for key in keys {
310 let value = rows
311 .first()?
312 .get(key)
313 .or_else(|| rows.first()?.get(format!("{}{}", LOCATION_KEY_PREFIX, key)))?;
314 conditions.push(match value {
315 serde_json::Value::Null => format!("{}=is.null", urlencode(key)),
316 serde_json::Value::String(text) => {
317 format!("{}=eq.{}", urlencode(key), urlencode(text))
318 }
319 other => format!("{}=eq.{}", urlencode(key), urlencode(&other.to_string())),
320 });
321 }
322
323 for row in rows.iter_mut() {
324 if let Some(object) = row.as_object_mut() {
325 for key in keys {
326 object.remove(&format!("{}{}", LOCATION_KEY_PREFIX, key));
327 }
328 }
329 }
330
331 Some(format!("/{}?{}", table, conditions.join("&")))
332}
333
334/// Percent-encode everything a query-string value may not carry literally.
335fn urlencode(value: &str) -> String {
336 let mut encoded = String::with_capacity(value.len());
337 for byte in value.as_bytes() {
338 match byte {
339 b'A'..=b'Z' | b'a'..=b'z' | b'0'..=b'9' | b'-' | b'_' | b'.' | b'~' => {
340 encoded.push(*byte as char)
341 }
342 other => encoded.push_str(&format!("%{:02X}", other)),
343 }
344 }
345 encoded
346}
347
348/// Whether this request must carry a body.
349///
350/// Insert, update and upsert all write values that can only come from one.
351/// A delete names its rows in the query string, and a function call may take
352/// no arguments at all.
353fn needs_payload(api_request: &ApiRequest) -> bool {
354 use postrust_core::api_request::{Action, DbAction, Mutation};
355
356 matches!(
357 &api_request.action,
358 Action::Db(DbAction::RelationMut {
359 mutation: Mutation::Create | Mutation::Update | Mutation::SingleUpsert,
360 ..
361 })
362 )
363}
364
365/// Classify a JWT failure the way PostgREST reports it.
366///
367/// A token that could not be read at all and one that was read and found
368/// wanting are different answers to the client: the first says the credential
369/// is unusable, the second says what about it was unacceptable. PostgREST
370/// gives them different codes, and clients branch on them.
371fn jwt_error(error: postrust_auth::JwtError) -> postrust_core::Error {
372 use postrust_auth::JwtError;
373
374 match error {
375 JwtError::NoIdentity => postrust_core::Error::MissingAuth,
376 JwtError::SecretMissing => postrust_core::Error::JwtSecretMissing,
377 // Read, and found wanting: the token itself is intact, and what it
378 // claims is what was not accepted.
379 JwtError::Claim(fault) => postrust_core::Error::JwtClaim(fault.to_string()),
380 JwtError::InvalidSignature => {
381 postrust_core::Error::InvalidJwt("JWT cryptographic operation failed".into())
382 }
383 JwtError::InvalidHeaderFormat => {
384 postrust_core::Error::InvalidJwt("Unsupported token type".into())
385 }
386 JwtError::NoSuitableKey => postrust_core::Error::NoSuitableJwtKey,
387 }
388}
389
390/// Execute an action plan.
391async fn execute_plan(
392 state: &AppState,
393 api_request: &ApiRequest,
394 plan: &ActionPlan,
395 auth: &postrust_auth::AuthResult,
396 added_join_columns: &[String],
397) -> Result<QueryResult, postrust_core::Error> {
398 match plan {
399 ActionPlan::Db(db_plan) => {
400 // A media type the schema renders itself replaces the whole
401 // response body: the aggregate is applied over the rows the
402 // request selected, so it sees exactly what would otherwise have
403 // been serialised as JSON.
404 let media_handler = {
405 let schema_cache = state.schema_cache().await;
406 read_target(api_request, Some(db_plan), &schema_cache).and_then(|qi| {
407 api_request.accept_media_types.iter().find_map(|media| {
408 schema_cache
409 .media_handler(
410 &api_request.schema,
411 media.content_type(),
412 &qi,
413 has_default_select(&api_request.query_params.select),
414 )
415 .map(|(handler, resolved)| {
416 (
417 resolved.to_string(),
418 format!(
419 "{}.{}",
420 postrust_sql::escape_ident(&handler.aggregate.schema),
421 postrust_sql::escape_ident(&handler.aggregate.name)
422 ),
423 // Applied to the table's own row, not to
424 // the derived one, whenever the derived
425 // row would be a worse description of it.
426 // A handler naming a table needs the row
427 // type by name; one taking `anyelement`
428 // needs the columns to still *be* what
429 // they are -- a handler calling
430 // `ST_AsGeoJSON` cannot find a geometry in
431 // a row whose geometry this side already
432 // rendered to jsonb. Only where the
433 // request did not reshape the rows, since
434 // then the two are the same rows.
435 handler.table.is_some()
436 || has_default_select(&api_request.query_params.select),
437 handler.base_type.clone(),
438 )
439 })
440 })
441 })
442 };
443
444 // Content negotiation. A media type is rendered by a builtin, by a
445 // handler the schema declared, or by the function's own return
446 // type -- and a request that names only types nothing here can
447 // render has not been answered by serving it JSON under a
448 // `Content-Type` it never asked for. PostgREST refuses it, naming
449 // everything that was asked for.
450 if media_handler.is_none()
451 && declared_media_type(db_plan, api_request).is_none()
452 && !api_request
453 .accept_media_types
454 .iter()
455 .any(renders_without_a_handler)
456 {
457 return Err(postrust_core::Error::InvalidMediaType(
458 api_request
459 .accept_media_types
460 .iter()
461 .map(postrust_core::api_request::MediaType::to_mime)
462 .collect::<Vec<_>>()
463 .join(", "),
464 ));
465 }
466
467 // Carry the parent's whole row out of the inner query when a
468 // computed relationship needs it as an argument. It has to be
469 // taken here, where the real table is still in scope: a bare
470 // reference to the table yields its composite type, whereas the
471 // derived table it becomes one level up yields a `record`.
472 let needs_parent_row = added_join_columns.iter().any(|c| c == PARENT_ROW_COLUMN)
473 || matches!(&media_handler, Some((_, _, true, _)));
474 let db_plan = &if needs_parent_row {
475 let mut adjusted = db_plan.clone();
476 // A function's result is a relation too, so the row is
477 // reachable there by exactly the same means -- which is what
478 // makes a computed relationship work on
479 // `/rpc/getallvideogames` and not only on a table.
480 //
481 // Named after the table it returns, not after the function.
482 // The builder aliases the call to that table (`FROM
483 // test.getallvideogames() AS videogames`), because a computed
484 // *column* has to be able to name the row it is a function of.
485 // Naming the function here instead asks for a relation that is
486 // no longer in scope, and the request that used to work --
487 // `?select=name,designer:computed_designers(name)` -- answers
488 // `column "getallvideogames" does not exist`.
489 let (tree, relation) = match &mut adjusted {
490 DbActionPlan::Read(tree) => {
491 let relation = tree.root.from.name.clone();
492 (Some(tree), relation)
493 }
494 DbActionPlan::Call { read, .. } => {
495 let relation = read
496 .as_ref()
497 .map(|tree| tree.root.from.name.clone())
498 .unwrap_or_default();
499 (read.as_mut(), relation)
500 }
501 // The row is taken inside the statement, by `RETURNING`,
502 // and read back off the CTE as an ordinary column -- so
503 // unlike the two above there is no relation to name here,
504 // and the read tree selects the column rather than a row.
505 DbActionPlan::MutateRead { mutate, read } => {
506 let returning = match mutate {
507 postrust_core::plan::MutatePlan::Insert { returning, .. }
508 | postrust_core::plan::MutatePlan::Update { returning, .. }
509 | postrust_core::plan::MutatePlan::Delete { returning, .. } => {
510 returning
511 }
512 };
513 if !returning.iter().any(|c| c == PARENT_ROW_COLUMN) {
514 returning.push(PARENT_ROW_COLUMN.to_string());
515 }
516 // A mutation that returns no body has no read tree,
517 // and nothing to embed into either.
518 if let Some(read) = read.as_mut() {
519 if !read.root.select.iter().any(|field| {
520 field.alias.as_deref() == Some(PARENT_ROW_COLUMN)
521 || field.field.name == PARENT_ROW_COLUMN
522 }) {
523 read.root.select.push(
524 postrust_core::plan::CoercibleSelectField::simple(
525 PARENT_ROW_COLUMN,
526 "",
527 ),
528 );
529 }
530 }
531 (None, String::new())
532 }
533 };
534
535 if let Some(tree) = tree {
536 // An empty select means every column; naming the row would
537 // otherwise replace them rather than join them.
538 if tree.root.select.is_empty() {
539 tree.root
540 .select
541 .push(postrust_core::plan::CoercibleSelectField::simple("*", ""));
542 }
543
544 let mut field =
545 postrust_core::plan::CoercibleField::simple(relation, String::new());
546 field.full_row = true;
547 tree.root
548 .select
549 .push(postrust_core::plan::CoercibleSelectField {
550 field,
551 aggregate: None,
552 aggregate_cast: None,
553 cast: None,
554 alias: Some(PARENT_ROW_COLUMN.to_string()),
555 });
556 }
557 adjusted
558 } else {
559 db_plan.clone()
560 };
561
562 // The `Location` of a created row is built from its primary key,
563 // which the client may not have selected. The columns are added
564 // under names of our own so they cannot collide with anything the
565 // request asked for, and are taken back out of the response once
566 // the header is built.
567 //
568 // Only for `Prefer: return=headers-only`, which is the request
569 // that has no other way to learn where the row went. A client
570 // given the row back can read the key out of it, and PostgREST
571 // sends no `Location` there -- so neither does this. Deciding it
572 // here also means the key columns are never added to a select
573 // that had no use for them.
574 let location_keys: Vec<String> = match db_plan {
575 DbActionPlan::MutateRead {
576 mutate: postrust_core::plan::MutatePlan::Insert { pk_cols, .. },
577 ..
578 } if !is_upsert(api_request) && wants_location(api_request) => pk_cols.clone(),
579 _ => Vec::new(),
580 };
581 // Whether the write created a row or merged into an existing one.
582 // The statement is the only place that can say, so the answer is
583 // carried out of it as a column and taken back out of the response.
584 let reports_inserted = matches!(
585 db_plan,
586 DbActionPlan::MutateRead {
587 mutate: postrust_core::plan::MutatePlan::Insert {
588 on_conflict: Some(_),
589 reports_inserted: true,
590 ..
591 },
592 ..
593 }
594 );
595
596 let db_plan = &if location_keys.is_empty() && !reports_inserted {
597 db_plan.clone()
598 } else {
599 let mut adjusted = db_plan.clone();
600 if let DbActionPlan::MutateRead {
601 read: Some(tree), ..
602 } = &mut adjusted
603 {
604 for key in &location_keys {
605 tree.root.select.push(
606 postrust_core::plan::CoercibleSelectField::with_alias(
607 key,
608 "",
609 &format!("{}{}", LOCATION_KEY_PREFIX, key),
610 ),
611 );
612 }
613 if reports_inserted {
614 tree.root
615 .select
616 .push(postrust_core::plan::CoercibleSelectField::simple(
617 postrust_core::query::INSERTED_COLUMN,
618 "boolean",
619 ));
620 }
621 }
622 adjusted
623 };
624
625 // Build SQL
626 let query = postrust_core::query::build_query(
627 &ActionPlan::Db(db_plan.clone()),
628 Some(&auth.role),
629 )?;
630
631 if !query.has_main() {
632 return Ok(QueryResult::default());
633 }
634
635 let (mut sql, params) = query.build_main();
636
637 // A `Prefer: count` needs the same query without its page: the
638 // total is what the filters match, not what this page returned.
639 // LIMIT and OFFSET render as literals, so dropping them leaves the
640 // placeholders and their numbering untouched and the count query
641 // can be bound with the very same parameters.
642 let mut count_sql = match api_request.preferences.count {
643 Some(_) => unpaged_sql(db_plan, &auth.role)?,
644 None => None,
645 };
646
647 // Embed related resources in the same query.
648 //
649 // Each relation becomes a correlated subselect in the SELECT list,
650 // so PostgreSQL builds the nested JSON while it already has the
651 // parent row -- one round trip for the whole tree instead of one per
652 // relation per level, and no grouping or attaching in this process.
653 //
654 // Parent columns stay ordinary typed columns, so they are converted
655 // to JSON by the same code as a request without embeds and a NUMERIC
656 // or timestamp renders identically either way.
657 let mut embed_filters = EmbedFilters {
658 filters: &api_request.query_params.filters,
659 orders: &api_request.query_params.order,
660 ranges: &api_request.query_params.ranges,
661 logic: &api_request.query_params.logic,
662 params: Vec::new(),
663 base: params.len(),
664 max_rows: api_request.max_rows,
665 alias_counter: 0,
666 };
667
668 // The single-query form cannot express a spread, whose columns have
669 // to land in the parent object rather than under a key of their
670 // own. One anywhere in the tree sends the whole request down the
671 // two-query path, which can: taking the single-query path for the
672 // plain relations would embed those and drop the spread, since the
673 // two-query path only runs when the first produced nothing.
674 let embed_parent = {
675 let schema_cache = state.schema_cache().await;
676 read_target(api_request, Some(db_plan), &schema_cache)
677 };
678 let embed_level = match embed_parent.clone() {
679 Some(parent_qi)
680 if api_request.query_params.select.iter().any(|item| {
681 matches!(
682 item,
683 postrust_core::api_request::SelectItem::Relation { .. }
684 | postrust_core::api_request::SelectItem::SpreadRelation { .. }
685 )
686 }) =>
687 {
688 let schema_cache = state.schema_cache().await;
689 // A computed relationship takes the parent's row, which
690 // now reaches here on every path: a read names the table,
691 // and a mutation returns the row out of the statement
692 // where it still has the table's type rather than the
693 // CTE's anonymous one.
694 let parent_row = PARENT_ROW_COLUMN_REF;
695 build_embed_expressions(
696 &schema_cache,
697 &parent_qi,
698 "src",
699 parent_row,
700 &api_request.query_params.select,
701 &mut embed_filters,
702 &[],
703 &[],
704 )?
705 }
706 _ => EmbedLevel::default(),
707 };
708 let mut embed_level = embed_level;
709
710 // Ordering by an embedded column is resolved here rather than in
711 // the read plan: the plan has no way to reach another table.
712 if let (Some(parent_qi), DbActionPlan::Read(tree)) = (&embed_parent, db_plan) {
713 let schema_cache = state.schema_cache().await;
714 embed_level.orders = build_embed_orders(
715 &schema_cache,
716 parent_qi,
717 &tree.root.order,
718 &api_request.query_params.select,
719 &mut embed_filters,
720 )?;
721 }
722
723 // Filter parameters are appended in the order the predicates were
724 // renumbered, so placeholder N in the wrapped SQL lines up with
725 // params[N - 1].
726 let mut params = params;
727 params.extend(embed_filters.params);
728
729 // Built whenever there were relations at all, not only when they
730 // produced expressions: `clients!inner()` contributes no column
731 // but does narrow the parent, and its predicate binds parameters
732 // that have already been appended -- left unapplied, the query is
733 // sent with more parameters than it uses.
734 if embed_level.saw_relations {
735 // `!inner` removes parent rows, so it has to be applied before
736 // the page is taken. Left where it is, the inner query's
737 // LIMIT would run first and the join would then trim an
738 // already-truncated page -- a request for one row could come
739 // back empty. So the inner query is rebuilt unpaged and the
740 // range moves out to the wrapper, after the join.
741 //
742 // Rebuilding is safe for parameter numbering: LIMIT and OFFSET
743 // render as literals, so dropping them leaves the placeholders
744 // and their count untouched.
745 //
746 // Ordering by an embedded resource's column needs the same
747 // treatment for the same reason: the value being ordered on is
748 // only reachable out here, so the page cannot be taken before
749 // it is known.
750 let orders_by_embed = matches!(db_plan, DbActionPlan::Read(tree)
751 if tree.root.order.iter().any(|t| t.relation.is_some()));
752
753 let mut page = None;
754 if !embed_level.inner_joins.is_empty() || orders_by_embed {
755 if let DbActionPlan::Read(tree) = db_plan {
756 let mut unpaged = tree.clone();
757 page = Some(std::mem::take(&mut unpaged.root.range));
758 sql = postrust_core::query::build_query(
759 &ActionPlan::Db(DbActionPlan::Read(unpaged)),
760 Some(&auth.role),
761 )?
762 .build_main()
763 .0;
764 }
765 }
766
767 let mut projection = String::from("src.*");
768 for (field_name, expression) in &embed_level.expressions {
769 projection.push_str(", ");
770 projection.push_str(expression);
771 projection.push_str(" AS ");
772 projection.push_str(&postrust_sql::escape_ident(field_name));
773 }
774 // A data-modifying `WITH` has to stay at the top level:
775 // PostgreSQL refuses one nested inside a subquery, which is
776 // where wrapping the whole statement would put it. The clause
777 // is lifted out and put back in front of the wrapper.
778 let inner_sql = std::mem::take(&mut sql);
779 let (with_clause, inner_sql) = split_leading_cte(&inner_sql);
780 sql = format!(
781 "{}SELECT {} FROM ({}) AS src",
782 with_clause, projection, inner_sql
783 );
784
785 // `?clients=is.null` asks whether the embed matched anything,
786 // not about a column. The embed's expression is the thing to
787 // test, and it is only in scope out here -- referring to the
788 // name it is given would be referring to a column of the very
789 // select that defines it.
790 let mut predicates = embed_level.inner_joins.clone();
791 for filter in &api_request.query_params.filters_root {
792 let Some((_, exists)) = embed_level
793 .filterable
794 .iter()
795 .find(|(name, _)| name == &filter.field.name)
796 else {
797 continue;
798 };
799 let negated = match &filter.op_expr.operation {
800 postrust_core::api_request::Operation::Is(
801 postrust_core::api_request::IsValue::Null,
802 ) => filter.op_expr.negated,
803 postrust_core::api_request::Operation::Is(
804 postrust_core::api_request::IsValue::NotNull,
805 ) => !filter.op_expr.negated,
806 _ => continue,
807 };
808 // `is.null` asks for parents the embed did not match, so
809 // the existence test is the negation of it.
810 predicates.push(match negated {
811 true => exists.clone(),
812 false => format!("NOT {}", exists),
813 });
814 }
815
816 // `or=(clientinfo.not.is.null,contact.not.is.null)` asks the
817 // same question of several embeds at once, and the answer is
818 // one predicate combining their existence tests. The read
819 // plan left the tree alone precisely so that it could be
820 // built here, where the embeds are in scope.
821 let embeds =
822 postrust_core::api_request::embedded_names(&api_request.query_params.select);
823 for (path, tree) in &api_request.query_params.logic {
824 if !path.is_empty() || !tree.names_only(&embeds) {
825 continue;
826 }
827 if let Some(predicate) = embed_logic_predicate(tree, &embed_level.filterable) {
828 predicates.push(predicate);
829 }
830 }
831
832 if !predicates.is_empty() {
833 sql.push_str(" WHERE ");
834 sql.push_str(&predicates.join(" AND "));
835 }
836
837 if !embed_level.orders.is_empty() {
838 sql.push_str(" ORDER BY ");
839 sql.push_str(&embed_level.orders.join(", "));
840 }
841
842 // `!inner` and `?rel=is.null` decide which parent rows survive,
843 // so the count has to be taken after them -- the same wrapper,
844 // over the unpaged query.
845 count_sql = count_sql.map(|base| {
846 let (with_clause, base) = split_leading_cte(&base);
847 let mut counted = format!(
848 "{}SELECT {} FROM ({}) AS src",
849 with_clause, projection, base
850 );
851 if !predicates.is_empty() {
852 counted.push_str(" WHERE ");
853 counted.push_str(&predicates.join(" AND "));
854 }
855 counted
856 });
857
858 if let Some(range) = page {
859 if let Some(limit) = range.limit {
860 sql.push_str(&format!(" LIMIT {}", limit));
861 }
862 if range.offset > 0 {
863 sql.push_str(&format!(" OFFSET {}", range.offset));
864 }
865 }
866 }
867
868 // `application/geo+json` has a rendering of its own even where no
869 // schema declares one: a FeatureCollection whose geometry is the
870 // table's geometry column and whose properties are the rest. A
871 // schema-declared handler overrides it, which is why this only
872 // runs when none matched.
873 //
874 // Asking for GeoJSON from rows that carry no geometry is an error
875 // rather than a plain JSON body: the client asked for features,
876 // and a list of objects with no geometry is not features.
877 let wants_geojson = media_handler.is_none()
878 && api_request
879 .accept_media_types
880 .iter()
881 .any(|m| matches!(m, postrust_core::api_request::MediaType::GeoJson));
882 let geojson_column = if wants_geojson {
883 let schema_cache = state.schema_cache().await;
884 let target = read_target(api_request, Some(db_plan), &schema_cache);
885 match target.as_ref().and_then(|qi| schema_cache.get_table(qi)) {
886 Some(table) => match geometry_key(table, &api_request.query_params.select) {
887 Some(column) => Some(column),
888 None => return Err(postrust_core::Error::MissingGeometryColumn),
889 },
890 // Nothing this side can name a geometry column on -- a
891 // scalar function, say. Left to the database, which is
892 // where the answer was coming from anyway.
893 None => None,
894 }
895 } else {
896 None
897 };
898
899 if let Some((media_type, aggregate, over_row, _)) = &media_handler {
900 // The aggregate takes the table's own row type. A derived
901 // table yields `record`, which will not cast to it, so where
902 // the handler names a table the aggregate is applied to the
903 // column carrying the real row -- the same column a computed
904 // relationship reads. A handler taking `anyelement` has no
905 // such requirement and takes the derived row directly.
906 let argument = if *over_row {
907 format!(
908 "pgrst_media.{}",
909 postrust_sql::escape_ident(PARENT_ROW_COLUMN)
910 )
911 } else {
912 "pgrst_media".to_string()
913 };
914 // `count(*)` alongside the handler: both are aggregates over
915 // the same set, and the rendered body is one value that has
916 // forgotten how many rows went into it. `Content-Range`
917 // describes the row set rather than the rendering, so the
918 // count has to be carried out of the same statement -- asking
919 // again afterwards would be a second snapshot.
920 let (with_clause, inner) = split_leading_cte(&sql);
921 sql = format!(
922 "{}SELECT {}({})::text AS pgrst_body, count(*) AS {} \
923 FROM ({}) pgrst_media",
924 with_clause, aggregate, argument, MEDIA_ROW_COUNT_COLUMN, inner
925 );
926 debug!("Rendering {} via {}", media_type, aggregate);
927 }
928
929 debug!("Executing SQL: {}", sql);
930 debug!("With {} parameters", params.len());
931
932 // Everything for this request runs in one transaction.
933 //
934 // `SET LOCAL ROLE` and `set_config(..., true)` are scoped to a
935 // transaction. Sent on a bare pooled connection they apply to their
936 // own implicit single-statement transaction and are gone before the
937 // next statement runs, so the query executed as the pool's login
938 // role with no JWT claims set -- row-level security and role grants
939 // were not being applied at all. PostgreSQL says so in its log:
940 // "SET LOCAL can only be used in transaction blocks".
941 //
942 // Embedding runs on this same transaction, which also means a
943 // parent row and its children come from one snapshot rather than
944 // from two separate reads.
945 let mut tx = state
946 .pool
947 .begin()
948 .await
949 .map_err(|e| postrust_core::Error::ConnectionPool(e.to_string()))?;
950
951 // A read request runs in a read-only transaction.
952 //
953 // Without this a GET can change data: `GET /rpc/some_volatile_fn`
954 // happily executes whatever the function does. PostgreSQL is the
955 // right place to enforce it, since it covers anything reachable
956 // from the statement -- a volatile function, a trigger, a rule --
957 // rather than only what the planner thought to look at.
958 if is_read_only(api_request) {
959 sqlx::query("SET TRANSACTION READ ONLY")
960 .execute(&mut *tx)
961 .await
962 .map_err(map_sqlx_error)?;
963 }
964
965 // Everything the request tells the database about itself, in
966 // one statement.
967 //
968 // These are `SET LOCAL` by another spelling -- `set_config(k, v,
969 // true)` -- and a function or an RLS policy reads them back with
970 // `current_setting`. Sent as one `SELECT` rather than one
971 // statement each, which is a round trip per setting otherwise, and
972 // there are eight of them before a request has done any work.
973 //
974 // The order is PostgREST's: the settings a role carries, then the
975 // role, then everything about the request. The role goes on after
976 // its own settings so that a `GRANT SET ON PARAMETER` on the
977 // authenticator still applies when they are set.
978 {
979 let mut path = vec![api_request.schema.clone()];
980 path.extend(state.config.db_extra_search_path.iter().cloned());
981 let search_path = path
982 .iter()
983 .map(|s| postrust_sql::escape_ident(s))
984 .collect::<Vec<_>>()
985 .join(", ");
986
987 // The claims carry the role, as PostgREST's do: a policy
988 // reading `request.jwt.claims` finds the role it is running as
989 // whether or not the token named one.
990 let mut claims: serde_json::Map<String, serde_json::Value> = auth
991 .claims
992 .iter()
993 .map(|(k, v)| (k.clone(), v.clone()))
994 .collect();
995 claims.insert(
996 "role".to_string(),
997 serde_json::Value::String(auth.role.clone()),
998 );
999
1000 let mut settings: Vec<(String, String)> = vec![
1001 ("search_path".to_string(), search_path),
1002 ("role".to_string(), auth.role.clone()),
1003 (
1004 "request.jwt.claims".to_string(),
1005 serde_json::Value::Object(claims).to_string(),
1006 ),
1007 ("request.method".to_string(), api_request.method.clone()),
1008 ("request.path".to_string(), api_request.path.clone()),
1009 (
1010 "request.headers".to_string(),
1011 json_object(&api_request.headers),
1012 ),
1013 (
1014 "request.cookies".to_string(),
1015 json_object(&api_request.cookies),
1016 ),
1017 ];
1018
1019 // `Prefer: timezone` is applied to the session, so every value
1020 // the database renders -- and every `now()` a function calls --
1021 // sees it. An unknown zone is PostgreSQL's to reject, and it
1022 // does, with a message naming the value the client sent.
1023 if let Some(timezone) = &api_request.preferences.timezone {
1024 settings.push(("timezone".to_string(), timezone.clone()));
1025 }
1026
1027 let calls = (0..settings.len())
1028 .map(|i| format!("set_config(${}, ${}, true)", i * 2 + 1, i * 2 + 2))
1029 .collect::<Vec<_>>()
1030 .join(", ");
1031 let sql = format!("SELECT {}", calls);
1032 let mut query = sqlx::query(&sql);
1033 for (name, value) in &settings {
1034 query = query.bind(name).bind(value);
1035 }
1036 query.execute(&mut *tx).await.map_err(map_sqlx_error)?;
1037 }
1038
1039 // Execute main query with bound parameters
1040 let rows = bind_params(sqlx::query(&sql), ¶ms)
1041 .fetch_all(&mut *tx)
1042 .await
1043 .map_err(|e| {
1044 error!("Query error: {}", e);
1045 map_sqlx_error(e)
1046 })?;
1047
1048 // Convert rows to JSON.
1049 //
1050 // `into_iter` matters for large result sets: each row's buffers are
1051 // freed as soon as it has been converted, rather than the whole
1052 // `Vec<PgRow>` staying alive alongside the whole `Vec<Value>`.
1053 // A rendered media type is a single value, not a row set: the
1054 // aggregate already produced the entire body.
1055 if let Some((media_type, base_type)) = media_handler
1056 .as_ref()
1057 .map(|(m, _, _, base)| (m.clone(), base.clone()))
1058 .or_else(|| declared_media_type(db_plan, api_request))
1059 {
1060 // The value's type is a domain this side has no mapping for,
1061 // so it arrives as PostgreSQL's text rendering of it. For
1062 // anything text-like that rendering *is* the value; for a
1063 // `bytea` it is `\x` and hex, which describes the bytes
1064 // rather than being them.
1065 let mut body = rows
1066 .first()
1067 .and_then(|row| raw_column(row, &base_type))
1068 .unwrap_or_default();
1069
1070 // PostgREST reads a handler's value as raw bytes in binary
1071 // format, and a `jsonb` in binary format begins with a version
1072 // byte -- which it forwards to the client. Its own suite
1073 // records the result and marks it: "TODO SOH (start of
1074 // heading) is being added to results".
1075 //
1076 // It is a bug, and the byte makes the body invalid JSON, so it
1077 // is not something to do by default. Compatibility mode is
1078 // where this server sets out to be PostgREST rather than to be
1079 // right, and a client written against PostgREST's bytes gets
1080 // PostgREST's bytes.
1081 if state.config.compat_mode && base_type == "jsonb" {
1082 body.insert(0, 0x01);
1083 }
1084
1085 // A rendered body is still a window on a result set, and
1086 // PostgREST reports one. The row count comes from the
1087 // aggregate's own statement where a handler rendered the rows;
1088 // where the value is a function's media-type domain there was
1089 // no aggregate, and the rows in hand are the answer.
1090 let rendered_rows = match media_handler.is_some() {
1091 true => rows
1092 .first()
1093 .and_then(|row| {
1094 sqlx::Row::try_get::<i64, _>(row, MEDIA_ROW_COUNT_COLUMN).ok()
1095 })
1096 .unwrap_or_default(),
1097 false => rows.len() as i64,
1098 };
1099 let total = match (&api_request.preferences.count, &count_sql) {
1100 (Some(preference), Some(count_sql)) => {
1101 resolve_count(
1102 &mut tx,
1103 preference,
1104 count_sql,
1105 ¶ms,
1106 api_request.max_rows,
1107 )
1108 .await?
1109 }
1110 _ => None,
1111 };
1112 let content_range = postrust_response::ContentRange::from_pagination(
1113 top_level_offset(api_request),
1114 rendered_rows,
1115 total,
1116 );
1117
1118 tx.commit()
1119 .await
1120 .map_err(|e| postrust_core::Error::ConnectionPool(e.to_string()))?;
1121 return Ok(QueryResult {
1122 status: StatusCode::OK,
1123 raw_body: Some((media_type, body)),
1124 content_range: Some(content_range),
1125 ..QueryResult::default()
1126 });
1127 }
1128
1129 let mut json_rows: Vec<serde_json::Value> = rows
1130 .into_iter()
1131 .map(|row| postrust_core::row_json::row_to_json(&row))
1132 .collect();
1133
1134 // Embeds already came back with the parent query when the SELECT
1135 // list carried relations. This path remains for anything the
1136 // single-query form did not handle.
1137 let two_query_parent = {
1138 let schema_cache = state.schema_cache().await;
1139 read_target(api_request, Some(db_plan), &schema_cache)
1140 };
1141 if let Some(parent_qi) = two_query_parent.filter(|_| !embed_level.saw_relations) {
1142 let schema_cache = state.schema_cache().await;
1143 embed_relations(
1144 &mut tx,
1145 &schema_cache,
1146 &parent_qi,
1147 &api_request.query_params.select,
1148 &mut json_rows,
1149 api_request.max_rows,
1150 )
1151 .await?;
1152 }
1153
1154 // A function can set the response's status and headers, and only
1155 // it knows whether it did -- so the settings are read back after
1156 // it has run, on the same transaction. Reads cannot set them
1157 // without a function of their own having run, which is what a
1158 // call or a mutation is.
1159 let (guc_headers, guc_status) =
1160 match is_mutation(db_plan) || matches!(db_plan, DbActionPlan::Call { .. }) {
1161 false => (None, None),
1162 true => {
1163 use sqlx::Row;
1164 let row = sqlx::query(
1165 "SELECT current_setting('response.headers', true), \
1166 current_setting('response.status', true)",
1167 )
1168 .fetch_one(&mut *tx)
1169 .await
1170 .map_err(map_sqlx_error)?;
1171 // Empty is absent. Once any transaction on this
1172 // connection has defined a custom setting, PostgreSQL
1173 // keeps the name for the rest of the session and
1174 // reports it as `''` rather than null -- so a request
1175 // that set nothing would otherwise inherit "the empty
1176 // headers" from whichever request last set some.
1177 let setting = |index: usize| {
1178 row.try_get::<Option<String>, _>(index)
1179 .ok()
1180 .flatten()
1181 .filter(|value| !value.trim().is_empty())
1182 };
1183 (setting(0), setting(1))
1184 }
1185 };
1186
1187 // Both are the function's own words, so both are checked before
1188 // they reach the response: a malformed one is a fault in the
1189 // schema, and saying so is more use than a header the client
1190 // cannot see or a status it cannot explain.
1191 if let Some(headers) = &guc_headers {
1192 if postrust_response::parse_guc_headers(headers).is_none() {
1193 return Err(postrust_core::Error::InvalidGucHeaders);
1194 }
1195 }
1196 let guc_status = match guc_status {
1197 None => None,
1198 Some(status) => Some(
1199 status
1200 .trim()
1201 .parse::<u16>()
1202 .ok()
1203 .and_then(|code| StatusCode::from_u16(code).ok())
1204 .ok_or(postrust_core::Error::InvalidGucStatus)?,
1205 ),
1206 };
1207
1208 // The count runs on the same transaction and so under the same
1209 // snapshot as the page it describes.
1210 let total = match (&api_request.preferences.count, &count_sql) {
1211 (Some(preference), Some(count_sql)) => {
1212 resolve_count(
1213 &mut tx,
1214 preference,
1215 count_sql,
1216 ¶ms,
1217 api_request.max_rows,
1218 )
1219 .await?
1220 }
1221 _ => None,
1222 };
1223
1224 // Reads take no locks worth holding and write plans commit their
1225 // work, so the transaction is committed either way.
1226 tx.commit()
1227 .await
1228 .map_err(|e| postrust_core::Error::ConnectionPool(e.to_string()))?;
1229
1230 // Drop columns that were only selected to join the embeds.
1231 //
1232 // Except where the embed answers to that very name: embedding
1233 // through a column -- `?select=client_id(*)` -- keys the result by
1234 // the column it joined on, and the embed is what the client asked
1235 // for. Stripping by name would take it along with the column.
1236 let embed_keys: std::collections::HashSet<&str> = embed_level
1237 .expressions
1238 .iter()
1239 .map(|(key, _)| key.as_str())
1240 .collect();
1241
1242 for column in added_join_columns
1243 .iter()
1244 .filter(|column| !embed_keys.contains(column.as_str()))
1245 {
1246 for row in json_rows.iter_mut() {
1247 if let Some(object) = row.as_object_mut() {
1248 object.remove(column);
1249 }
1250 }
1251 }
1252
1253 // A spread's columns belong to the object above it. This happens
1254 // after the join columns are dropped, not before: a spread may
1255 // legitimately produce a column of the same name as one added for
1256 // joining, and dropping it afterwards would take the wrong one.
1257 let spread_columns: std::collections::HashMap<String, Vec<String>> =
1258 embed_level.spread_columns.iter().cloned().collect();
1259 for row in json_rows.iter_mut() {
1260 flatten_spreads(row, &spread_columns);
1261 }
1262
1263 // In PostgREST-compatibility mode, reshape RPC responses to match
1264 // PostgREST: un-nest the function-name-keyed column and return a
1265 // bare value for non-set-returning functions.
1266 let (mut json_rows, singular) = if state.config.compat_mode {
1267 if let ActionPlan::Db(DbActionPlan::Call { call, .. }) = plan {
1268 unwrap_rpc_rows(json_rows, call)
1269 } else {
1270 (json_rows, false)
1271 }
1272 } else {
1273 (json_rows, false)
1274 };
1275
1276 // A function returning `void` has no result to report. PostgREST
1277 // answers 204 rather than a body of `null`, which is the honest
1278 // shape for something that returns nothing.
1279 let returns_void = matches!(
1280 db_plan,
1281 DbActionPlan::Call { call, .. } if call.returns_void
1282 );
1283
1284 // `Prefer: max-affected` guards against a filter that turned out
1285 // to match more than the client meant. Nothing has been committed
1286 // yet, so refusing here undoes the whole statement.
1287 if let (Some(limit), postrust_core::api_request::PreferHandling::Strict) = (
1288 api_request.preferences.max_affected,
1289 &api_request.preferences.handling,
1290 ) {
1291 let affected = json_rows.len() as i64;
1292 let writes = is_mutation(db_plan) || matches!(db_plan, DbActionPlan::Call { .. });
1293 if writes && affected > limit {
1294 return Err(postrust_core::Error::MaxAffectedExceeded(affected));
1295 }
1296 }
1297
1298 // Any row this statement created rather than merged into. Read
1299 // before the column is taken back out of the response.
1300 let created_a_row = reports_inserted
1301 && json_rows.iter().any(|row| {
1302 row.get(postrust_core::query::INSERTED_COLUMN)
1303 == Some(&serde_json::Value::Bool(true))
1304 });
1305 if reports_inserted {
1306 for row in json_rows.iter_mut() {
1307 if let Some(object) = row.as_object_mut() {
1308 object.remove(postrust_core::query::INSERTED_COLUMN);
1309 }
1310 }
1311 }
1312
1313 // A `PUT` writes the row its URL names, and exactly that row. The
1314 // filters were applied to the body, so a body naming some other
1315 // row wrote nothing -- and one naming several wrote several. Both
1316 // are the same mistake, and PostgreSQL cannot report it because
1317 // neither is a database error.
1318 if is_upsert(api_request) && json_rows.len() != 1 {
1319 return Err(postrust_core::Error::PutMatchingPk);
1320 }
1321
1322 // The created row's own address, taken from its key and then
1323 // removed from the body -- the client asked for a `?select=`, not
1324 // for the key.
1325 let location = build_location(api_request, &mut json_rows, &location_keys);
1326
1327 // A mutation returns the affected rows only when the caller asked
1328 // for them; otherwise the body is empty whatever the status.
1329 let omit_body = is_mutation(db_plan) && !wants_representation(api_request);
1330 // Taken before the rows are dropped: how many a write touched is
1331 // still the answer to `Content-Range` and to a request for a
1332 // single object.
1333 let affected = json_rows.len();
1334 let rows = if omit_body { Vec::new() } else { json_rows };
1335
1336 // `application/geo+json` is a rendering of these same rows, done
1337 // here rather than in SQL so that everything a mutation reports
1338 // has already happened to them: the status derived, the location
1339 // taken from the key, the bookkeeping columns removed. Built in
1340 // SQL it saw the row as the statement left it, and leaked
1341 // `pgrst_inserted` into a feature's properties.
1342 let geojson_body = geojson_column
1343 .as_ref()
1344 .filter(|_| !omit_body)
1345 .map(|column| {
1346 (
1347 "application/geo+json".to_string(),
1348 feature_collection(&rows, column),
1349 )
1350 });
1351
1352 // PostgREST reports the returned window on every successful data
1353 // response -- reads, mutations and RPC alike, but not on errors or
1354 // OPTIONS. Without an exact count the total is unknown, which
1355 // renders as `*` and keeps the status at 200.
1356 // The window starts where the request asked it to, and `offset`
1357 // overrides the `Range` header exactly as it does when the plan
1358 // resolves the range -- otherwise the reported window says 0 for a
1359 // request that plainly skipped rows.
1360 let offset = top_level_offset(api_request);
1361
1362 // What a mutation reports is not a window on a result set. An
1363 // insert or a delete reports no window at all -- `*` -- because
1364 // the rows it wrote are not a page of anything; an update reports
1365 // how many it changed; and an upsert reports nothing, the client
1366 // having named the row itself. PostgREST draws exactly these
1367 // lines, and a client reading the header expects them.
1368 use postrust_core::api_request::Mutation;
1369 let content_range = match mutation_kind(api_request) {
1370 Some(Mutation::SingleUpsert) => None,
1371 Some(Mutation::Create) | Some(Mutation::Delete) => {
1372 Some(postrust_response::ContentRange::new(1, 0, total))
1373 }
1374 Some(Mutation::Update) => Some(postrust_response::ContentRange::new(
1375 0,
1376 affected as i64 - 1,
1377 total,
1378 )),
1379 None => Some(postrust_response::ContentRange::from_pagination(
1380 offset,
1381 rows.len() as i64,
1382 total,
1383 )),
1384 };
1385
1386 // An offset past the end of the result is a range the server
1387 // cannot satisfy, and saying how many rows there actually are is
1388 // what lets the client correct it. Only reachable with a count
1389 // preference: without one the total is unknown and there is
1390 // nothing to be past the end of.
1391 if content_range
1392 .as_ref()
1393 .is_some_and(|range| range.status() == StatusCode::RANGE_NOT_SATISFIABLE)
1394 && mutation_kind(api_request).is_none()
1395 {
1396 return Err(postrust_core::Error::InvalidRange(format!(
1397 "An offset of {} was requested, but there are only {} rows.",
1398 offset,
1399 total.unwrap_or(0)
1400 )));
1401 }
1402
1403 Ok(QueryResult {
1404 // A status the function set outright wins: it knows things
1405 // about the outcome that the request's shape cannot say.
1406 status: match (guc_status, returns_void) {
1407 (Some(status), _) => status,
1408 (None, true) => StatusCode::NO_CONTENT,
1409 (None, false) => {
1410 mutation_status(db_plan, api_request, created_a_row, reports_inserted)
1411 .unwrap_or_else(|| {
1412 content_range
1413 .as_ref()
1414 .map(postrust_response::ContentRange::status)
1415 .unwrap_or(StatusCode::OK)
1416 })
1417 }
1418 },
1419 rows,
1420 singular,
1421 omit_body: omit_body || returns_void,
1422 affected,
1423 location,
1424 content_range,
1425 guc_headers,
1426 raw_body: geojson_body,
1427 ..Default::default()
1428 })
1429 }
1430 ActionPlan::Info(info_plan) => {
1431 use postrust_core::plan::InfoPlan;
1432
1433 // Return appropriate metadata based on the info type
1434 let response_data = match info_plan {
1435 InfoPlan::OpenApiSpec => {
1436 // Return basic server info for root endpoint
1437 serde_json::json!({
1438 "name": "postrust",
1439 "version": env!("CARGO_PKG_VERSION"),
1440 "description": "PostgREST-compatible REST API for PostgreSQL"
1441 })
1442 }
1443 InfoPlan::RelationInfo(qi) => {
1444 serde_json::json!({
1445 "schema": qi.schema,
1446 "name": qi.name,
1447 "type": "relation"
1448 })
1449 }
1450 InfoPlan::RoutineInfo(qi) => {
1451 serde_json::json!({
1452 "schema": qi.schema,
1453 "name": qi.name,
1454 "type": "routine"
1455 })
1456 }
1457 };
1458
1459 Ok(QueryResult {
1460 status: StatusCode::OK,
1461 rows: vec![response_data],
1462 ..Default::default()
1463 })
1464 }
1465 }
1466}
1467
1468/// Answer an `OPTIONS` with the methods its resource actually has.
1469///
1470/// This runs outside the CORS layer because that layer answers every `OPTIONS`
1471/// itself and never calls what it wraps -- `if parts.method == Method::OPTIONS`
1472/// returns immediately, preflight or not. So an `OPTIONS` never reached the
1473/// request handler at all: the body was empty because nothing built one, and
1474/// `Allow` was missing because the code that knows the answer was never asked.
1475///
1476/// The CORS layer still writes the response, and this adds to it, so a
1477/// preflight keeps every header it had.
1478pub async fn options_allow(
1479 State(state): State<Arc<AppState>>,
1480 request: Request,
1481 next: axum::middleware::Next,
1482) -> Response {
1483 if request.method() != http::Method::OPTIONS {
1484 return next.run(request).await;
1485 }
1486
1487 // Not every route is the API's. A layer added to the router wraps every
1488 // one of them, so `/admin` and `/_` arrive here too, and answering those
1489 // with what a table allows -- or with "no such table" -- takes a working
1490 // preflight and turns it into a 404.
1491 let Some(path) = api_path(request.uri().path(), state.config.compat_mode) else {
1492 return next.run(request).await;
1493 };
1494
1495 let verbatim_db_errors = state.config.compat_mode;
1496
1497 // Built here, from borrows, before anything is awaited: `axum::body::Body`
1498 // is `Send` but not `Sync`, and `&T` is `Send` only where `T` is `Sync`, so
1499 // a `&Request` held across an await makes the whole future not `Send` --
1500 // which a middleware has to be.
1501 let probe = probe_request(&request, path);
1502 let allow = match probe {
1503 Ok(probe) => resolve_allow(&state, probe).await,
1504 Err(error) => Err(error),
1505 };
1506
1507 let mut response = next.run(request).await;
1508
1509 match allow {
1510 Ok(allow) => {
1511 if let Ok(value) = http::HeaderValue::from_str(&allow) {
1512 response.headers_mut().insert(http::header::ALLOW, value);
1513 }
1514 response
1515 }
1516 // Asking what may be done with a table that is not there is answered
1517 // the same way as asking for the table.
1518 Err(error) => {
1519 let mut refusal = error_response(error, verbatim_db_errors).into_response();
1520 // The CORS layer wrote its headers on the response this replaces.
1521 // Without them a browser reports the request as blocked by CORS
1522 // rather than as the 404 it earned, which is a different thing to
1523 // go and fix.
1524 for (name, value) in response.headers() {
1525 if name.as_str().starts_with("access-control-") || name == http::header::VARY {
1526 refusal.headers_mut().append(name, value.clone());
1527 }
1528 }
1529 refusal
1530 }
1531 }
1532}
1533
1534/// The path the API parser should read, or `None` where the request names
1535/// something that is not the REST surface.
1536///
1537/// A layer added with `Router::layer` wraps each endpoint, and `nest` puts its
1538/// prefix-stripping *inside* that endpoint -- so the path seen here is the one
1539/// the client wrote and still carries the mount it arrived under, unlike the
1540/// one the handler reads.
1541fn api_path(path: &str, compat_mode: bool) -> Option<&str> {
1542 /// Whether a path names a mount point or something under it.
1543 fn under(path: &str, mount: &str) -> bool {
1544 path == mount || path.strip_prefix(mount).is_some_and(|r| r.starts_with('/'))
1545 }
1546
1547 // Every mount this server answers itself. In compatibility mode the REST
1548 // surface sits at the root, so a path not named here is read as a table:
1549 // `/healthz` becomes a question about a table called `healthz`, and the
1550 // two-segment ones are not a valid resource at all. Either way the
1551 // refusal replaces the preflight the CORS layer wrote, which is how a
1552 // browser comes to report a working endpoint as blocked.
1553 for mount in [
1554 "/admin",
1555 "/_",
1556 "/api/graphql",
1557 "/healthz",
1558 "/v1/version",
1559 "/v1/graphql",
1560 "/v1alpha1/graphql",
1561 ] {
1562 if under(path, mount) {
1563 return None;
1564 }
1565 }
1566 if under(path, "/api") {
1567 return Some(match path.strip_prefix("/api") {
1568 None | Some("") => "/",
1569 Some(rest) => rest,
1570 });
1571 }
1572 // Everywhere else the REST surface exists only in compatibility mode,
1573 // where it answers at the root. Without it `/` is this server's own
1574 // directory of endpoints, which allows what its handler allows.
1575 compat_mode.then_some(path)
1576}
1577
1578/// The request again, with an empty body and the API's own path, for the
1579/// parser to read.
1580fn probe_request(
1581 request: &Request,
1582 path: &str,
1583) -> Result<http::Request<bytes::Bytes>, postrust_core::Error> {
1584 let target = match request.uri().query() {
1585 Some(query) => format!("{path}?{query}"),
1586 None => path.to_string(),
1587 };
1588 let mut builder = http::Request::builder()
1589 .method(request.method().clone())
1590 .uri(target);
1591 for (name, value) in request.headers() {
1592 builder = builder.header(name, value);
1593 }
1594 builder
1595 .body(bytes::Bytes::new())
1596 .map_err(|e| postrust_core::Error::Internal(e.to_string()))
1597}
1598
1599/// What the resource this request names allows, or why it cannot say.
1600///
1601/// The path is resolved by the same parser the handler uses rather than by
1602/// splitting on `/` here, so a percent-encoded table name, an `Accept-Profile`
1603/// naming another schema, and the `/rpc/` prefix all mean what they mean
1604/// everywhere else.
1605async fn resolve_allow(
1606 state: &AppState,
1607 probe: http::Request<bytes::Bytes>,
1608) -> Result<String, postrust_core::Error> {
1609 use postrust_core::api_request::Action;
1610
1611 let api_request = parse_request(
1612 &probe,
1613 state.default_schema(),
1614 state.schemas(),
1615 state.config.db_max_rows,
1616 )?;
1617
1618 let schema_cache = state.schema_cache().await;
1619 match &api_request.action {
1620 Action::RelationInfo(qi) => Ok(relation_allow(schema_cache.require_table(qi)?)),
1621 Action::RoutineInfo { qi, .. } => Ok(routine_allow(schema_cache.get_routines(qi))),
1622 // The schema description, which is read and nothing else.
1623 _ => Ok("OPTIONS,GET,HEAD".to_string()),
1624 }
1625}
1626
1627/// The methods a relation answers, for the `Allow` of an OPTIONS.
1628///
1629/// Reading is always offered. Writing follows what the relation can actually
1630/// do, which for a view is what its triggers give it: a view with only an
1631/// INSERT trigger takes a POST and nothing else.
1632///
1633/// PUT is the exception that is not simply a permission. It replaces one row
1634/// named by its primary key, so a relation without a primary key cannot
1635/// answer it however writable it otherwise is -- which is why an auto-updatable
1636/// view without a key offers POST, PATCH and DELETE but not PUT.
1637///
1638fn relation_allow(table: &postrust_core::schema_cache::Table) -> String {
1639 let mut methods = vec!["OPTIONS", "GET", "HEAD"];
1640
1641 if table.insertable {
1642 methods.push("POST");
1643 }
1644 if table.insertable && table.updatable && !table.pk_cols.is_empty() {
1645 methods.push("PUT");
1646 }
1647 if table.updatable {
1648 methods.push("PATCH");
1649 }
1650 if table.deletable {
1651 methods.push("DELETE");
1652 }
1653
1654 methods.join(",")
1655}
1656
1657/// The methods a function answers, for the `Allow` of an OPTIONS.
1658///
1659/// A volatile function may change data, so it is reachable only by POST. One
1660/// that cannot can also be read, and is offered under GET and HEAD as well.
1661/// An overloaded name is readable when any of its overloads is.
1662fn routine_allow(routines: Option<&Vec<postrust_core::schema_cache::Routine>>) -> String {
1663 let readable = routines.is_some_and(|routines| {
1664 routines.iter().any(|routine| {
1665 !matches!(
1666 routine.volatility,
1667 postrust_core::schema_cache::FuncVolatility::Volatile
1668 )
1669 })
1670 });
1671
1672 match readable {
1673 true => "OPTIONS,GET,HEAD,POST".to_string(),
1674 false => "OPTIONS,POST".to_string(),
1675 }
1676}
1677
1678/// Reshape RPC rows for PostgREST-compatibility mode.
1679///
1680/// `SELECT * FROM func(...)` names its single output column after the function
1681/// (for scalar/`json` returns), which serializes to rows like
1682/// `[{"func": <value>}]`. PostgREST instead returns the bare value. This
1683/// un-nests that wrapper column and reports whether the result should be
1684/// rendered as a single (un-arrayed) value — i.e. when the function is not
1685/// set-returning.
1686///
1687/// The decision is driven by the plan's return-type metadata: composite and
1688/// `record` returns (`RETURNS TABLE`, row types) have real output columns and
1689/// are never un-nested, even when a single column happens to share the
1690/// function's name (e.g. `CREATE FUNCTION foo() RETURNS TABLE(foo int)`).
1691fn unwrap_rpc_rows(
1692 rows: Vec<serde_json::Value>,
1693 call: &CallPlan,
1694) -> (Vec<serde_json::Value>, bool) {
1695 let singular = !call.returns_set;
1696
1697 if call.returns_composite {
1698 return (rows, singular);
1699 }
1700
1701 let fname = call.function.name.as_str();
1702 let unwrapped = rows
1703 .into_iter()
1704 .map(|row| match row {
1705 serde_json::Value::Object(ref map) if map.len() == 1 && map.contains_key(fname) => {
1706 map.get(fname).cloned().unwrap_or(serde_json::Value::Null)
1707 }
1708 other => other,
1709 })
1710 .collect();
1711
1712 (unwrapped, singular)
1713}
1714
1715/// The table a read targets, if the request is a read.
1716fn read_target(
1717 api_request: &postrust_core::api_request::ApiRequest,
1718 db_plan: Option<&DbActionPlan>,
1719 schema_cache: &postrust_core::SchemaCache,
1720) -> Option<postrust_core::api_request::QualifiedIdentifier> {
1721 use postrust_core::api_request::{Action, DbAction};
1722
1723 match &api_request.action {
1724 Action::Db(DbAction::RelationRead { qi, .. }) => Some(qi.clone()),
1725 // The rows a mutation affected are rows of that table, so `?select=`
1726 // reaches through them to related resources exactly as it does on a
1727 // read -- `DELETE /tasks?select=name,project:projects(id)` says what
1728 // was deleted and what it belonged to.
1729 Action::Db(DbAction::RelationMut { qi, .. }) => Some(qi.clone()),
1730 // A function returning a table's rows embeds, and renders, exactly as
1731 // that table does.
1732 //
1733 // Which table that is was settled when the overload was chosen: a
1734 // name may carry several signatures returning different things, and
1735 // asking the cache again by name answers for whichever comes first.
1736 // It also knows the answer for a return type this side cannot read
1737 // off the name at all, such as a domain over a table.
1738 Action::Db(DbAction::Routine { qi, .. }) => match db_plan {
1739 Some(DbActionPlan::Call { read, .. }) => {
1740 read.as_ref().map(|tree| tree.root.from.clone())
1741 }
1742 _ => schema_cache
1743 .routine_returned_table(qi)
1744 .map(|table| table.qualified_identifier()),
1745 },
1746 _ => None,
1747 }
1748}
1749
1750/// One `?or=` or `?and=` over embedded resources, as a `WHERE` fragment.
1751///
1752/// Every leaf asks whether an embed matched, which is an existence test, and
1753/// the tree is what combines them. `None` where any leaf asks something else,
1754/// since half a condition is worse than none.
1755fn embed_logic_predicate(
1756 tree: &postrust_core::api_request::LogicTree,
1757 filterable: &[(String, String)],
1758) -> Option<String> {
1759 use postrust_core::api_request::{IsValue, LogicOperator, LogicTree, Operation};
1760
1761 match tree {
1762 LogicTree::Expr {
1763 negated,
1764 op,
1765 children,
1766 } => {
1767 let rendered = children
1768 .iter()
1769 .map(|child| embed_logic_predicate(child, filterable))
1770 .collect::<Option<Vec<String>>>()?;
1771 let joined = rendered.join(match op {
1772 LogicOperator::And => " AND ",
1773 LogicOperator::Or => " OR ",
1774 });
1775 Some(match negated {
1776 true => format!("NOT ({joined})"),
1777 false => format!("({joined})"),
1778 })
1779 }
1780 LogicTree::Stmt(filter) => {
1781 let (_, exists) = filterable
1782 .iter()
1783 .find(|(name, _)| name == &filter.field.name)?;
1784 // `is.null` asks for parents the embed did not match, so the
1785 // existence test is the negation of it.
1786 let negated = match &filter.op_expr.operation {
1787 Operation::Is(IsValue::Null) => filter.op_expr.negated,
1788 Operation::Is(IsValue::NotNull) => !filter.op_expr.negated,
1789 _ => return None,
1790 };
1791 Some(match negated {
1792 true => exists.clone(),
1793 false => format!("NOT {exists}"),
1794 })
1795 }
1796 }
1797}
1798
1799/// Round a geometry's numbers to nine decimal places.
1800///
1801/// Not cosmetic: nine is what `ST_AsGeoJSON` emits by default, and a geometry
1802/// read through the type's own rendering carries every digit the double has.
1803/// Nine decimal places of latitude is under a tenth of a millimetre, so the
1804/// digits past it describe floating-point representation rather than any
1805/// place on the earth.
1806fn round_coordinates(value: &mut serde_json::Value) {
1807 match value {
1808 serde_json::Value::Array(items) => items.iter_mut().for_each(round_coordinates),
1809 serde_json::Value::Object(fields) => fields.values_mut().for_each(round_coordinates),
1810 serde_json::Value::Number(number) => {
1811 let Some(float) = number.as_f64() else { return };
1812 if float.fract() == 0.0 {
1813 return;
1814 }
1815 // Through the decimal text rather than by arithmetic: scaling by a
1816 // power of ten and rounding back reintroduces exactly the digits
1817 // this is meant to drop.
1818 if let Ok(rounded) = format!("{float:.9}").parse::<f64>() {
1819 if let Some(rounded) = serde_json::Number::from_f64(rounded) {
1820 *number = rounded;
1821 }
1822 }
1823 }
1824 _ => {}
1825 }
1826}
1827
1828/// Where the response's window starts.
1829///
1830/// The same resolution the plan itself used, asked again rather than restated:
1831/// the JSON path and the rendered-media path both report a `Content-Range`,
1832/// and a `Content-Range` that disagrees with the rows underneath it is worse
1833/// than none at all.
1834fn top_level_offset(api_request: &ApiRequest) -> i64 {
1835 postrust_core::plan::resolve_top_level_range(api_request).offset
1836}
1837
1838/// The first column of a row, as the bytes the client should receive.
1839///
1840/// A media type the schema declared is carried by a domain, and a domain is a
1841/// type this side has no decoder for -- so the value arrives as PostgreSQL's
1842/// own text rendering. For text, xml or json that rendering is the value. For
1843/// `bytea` it is `\x` followed by hex, which is a description of the bytes
1844/// and not the bytes, so it is decoded back.
1845fn raw_column(row: &sqlx::postgres::PgRow, base_type: &str) -> Option<Vec<u8>> {
1846 use sqlx::{Row, ValueRef};
1847
1848 let value = row.try_get_raw(0).ok()?;
1849 if value.is_null() {
1850 return None;
1851 }
1852 let bytes = value.as_bytes().ok()?.to_vec();
1853 if base_type != "bytea" {
1854 return Some(bytes);
1855 }
1856 // Text or binary, depending on whether this connection asked for a format
1857 // it could name -- so the `\x` rendering is decoded where it appears and
1858 // the bytes are passed through where they do not.
1859 let Some(hex) = bytes.strip_prefix(b"\\x") else {
1860 return Some(bytes);
1861 };
1862 hex.chunks(2)
1863 .map(|pair| {
1864 let digits = std::str::from_utf8(pair).ok()?;
1865 u8::from_str_radix(digits, 16).ok()
1866 })
1867 .collect::<Option<Vec<u8>>>()
1868 .or(Some(bytes))
1869}
1870
1871/// Whether the selection is the one a table has when none was asked for.
1872///
1873/// A handler declared for a table renders the table's own rows, so it applies
1874/// only where the request did not reshape them. `?select=*` is that shape
1875/// written out; `?select=id` and any embed are not.
1876fn has_default_select(select: &[postrust_core::api_request::SelectItem]) -> bool {
1877 use postrust_core::api_request::SelectItem;
1878
1879 match select {
1880 [] => true,
1881 [SelectItem::Field { field, alias, .. }] => field.name == "*" && alias.is_none(),
1882 _ => false,
1883 }
1884}
1885
1886/// Whether this server renders the type itself, with nothing declared for it.
1887fn renders_without_a_handler(media: &postrust_core::api_request::MediaType) -> bool {
1888 use postrust_core::api_request::MediaType;
1889
1890 matches!(
1891 media,
1892 MediaType::ApplicationJson
1893 | MediaType::Any
1894 | MediaType::TextCsv
1895 | MediaType::GeoJson
1896 | MediaType::SingularJson { .. }
1897 | MediaType::ArrayJson { .. }
1898 | MediaType::OpenApi
1899 )
1900}
1901
1902/// Render rows as a GeoJSON `FeatureCollection`.
1903///
1904/// The geometry arrives already rendered: a user-defined type is rendered by
1905/// the database, and what PostGIS renders a geometry as is GeoJSON -- with a
1906/// `crs` key, which a feature's geometry does not carry. Everything else on
1907/// the row is the feature's properties, the geometry excepted, since a
1908/// feature's geometry is not also one of its properties.
1909fn feature_collection(rows: &[serde_json::Value], column: &str) -> Vec<u8> {
1910 let features: Vec<serde_json::Value> = rows
1911 .iter()
1912 .map(|row| {
1913 let mut properties = row.clone();
1914 let geometry = properties
1915 .as_object_mut()
1916 .and_then(|fields| fields.remove(column))
1917 .map(|mut geometry| {
1918 if let Some(fields) = geometry.as_object_mut() {
1919 fields.remove("crs");
1920 }
1921 round_coordinates(&mut geometry);
1922 geometry
1923 })
1924 .unwrap_or(serde_json::Value::Null);
1925 serde_json::json!({
1926 "type": "Feature",
1927 "geometry": geometry,
1928 "properties": properties,
1929 })
1930 })
1931 .collect();
1932 serde_json::to_vec(&serde_json::json!({
1933 "type": "FeatureCollection",
1934 "features": features,
1935 }))
1936 .unwrap_or_default()
1937}
1938
1939/// The key under which a GeoJSON feature's geometry is read.
1940///
1941/// A table may carry more than one geometry column, and `?select=` is what
1942/// says which of them the request is about: `select=id,name,range_area` asks
1943/// for a collection of areas, not for whichever geometry the catalogue
1944/// happens to list first. The name wanted is the one the response uses, so an
1945/// aliased column is found under its alias -- which is also the key that has
1946/// to come back out of `properties`, since a feature's geometry is not also
1947/// one of its properties.
1948///
1949/// `None` means the rows carry no geometry at all, which is not a GeoJSON
1950/// response with something missing from it but a request that cannot be
1951/// answered.
1952fn geometry_key(
1953 table: &postrust_core::schema_cache::Table,
1954 select: &[postrust_core::api_request::SelectItem],
1955) -> Option<String> {
1956 use postrust_core::api_request::SelectItem;
1957
1958 let is_geometry = |name: &str| {
1959 table
1960 .get_column(name)
1961 .is_some_and(|c| matches!(c.nominal_type.as_str(), "geometry" | "geography"))
1962 };
1963
1964 // `*` and a named column can both appear in one selection, and a column
1965 // named outright is the one meant even so: `select=*,range_area` is a
1966 // request about areas.
1967 let mut selects_every_column = select.is_empty();
1968 for item in select {
1969 if let SelectItem::Field { field, alias, .. } = item {
1970 if field.name == "*" {
1971 selects_every_column = true;
1972 } else if is_geometry(&field.name) {
1973 return Some(alias.clone().unwrap_or_else(|| field.name.clone()));
1974 }
1975 }
1976 }
1977
1978 if !selects_every_column {
1979 return None;
1980 }
1981 table
1982 .columns
1983 .values()
1984 .find(|c| matches!(c.nominal_type.as_str(), "geometry" | "geography"))
1985 .map(|c| c.name.clone())
1986}
1987
1988/// Whether any part of the selection, at any depth, asks for an aggregate.
1989fn selects_an_aggregate(api_request: &ApiRequest) -> bool {
1990 use postrust_core::api_request::SelectItem;
1991
1992 fn walk(items: &[SelectItem]) -> bool {
1993 items.iter().any(|item| match item {
1994 SelectItem::Field { aggregate, .. } => aggregate.is_some(),
1995 SelectItem::Relation { select, .. } => walk(select),
1996 SelectItem::SpreadRelation { select, .. } => walk(select),
1997 })
1998 }
1999
2000 walk(&api_request.query_params.select)
2001}
2002
2003/// Whether this request is a read, and so may run read-only.
2004///
2005/// It is the method that decides, not the plan: calling a function is a read
2006/// over GET and a write over POST, and the same function may be either. Asking
2007/// the plan instead would class every call as a read and refuse the writes a
2008/// POST is entitled to make.
2009fn is_read_only(api_request: &ApiRequest) -> bool {
2010 use postrust_core::api_request::{Action, DbAction, InvokeMethod};
2011
2012 match &api_request.action {
2013 Action::Db(DbAction::RelationRead { .. }) | Action::Db(DbAction::SchemaRead { .. }) => true,
2014 Action::Db(DbAction::Routine { invoke_method, .. }) => {
2015 matches!(invoke_method, InvokeMethod::InvRead { .. })
2016 }
2017 Action::Db(DbAction::RelationMut { .. }) => false,
2018 // Metadata only; nothing is executed against the database.
2019 Action::RelationInfo(_) | Action::RoutineInfo { .. } | Action::SchemaInfo => true,
2020 }
2021}
2022
2023/// The plan's SQL with its page removed, for counting what the filters match.
2024///
2025/// Only reads and calls are counted. A mutation's `Content-Range` reports the
2026/// rows it affected, which is the result set itself.
2027fn unpaged_sql(
2028 db_plan: &postrust_core::plan::DbActionPlan,
2029 role: &str,
2030) -> Result<Option<String>, postrust_core::Error> {
2031 use postrust_core::plan::{ActionPlan, DbActionPlan};
2032
2033 let unpaged = match db_plan {
2034 DbActionPlan::Read(tree) => {
2035 let mut tree = tree.clone();
2036 tree.root.range = Default::default();
2037 DbActionPlan::Read(tree)
2038 }
2039 DbActionPlan::Call { call, read } => {
2040 let read = read.as_ref().map(|tree| {
2041 let mut tree = tree.clone();
2042 tree.root.range = Default::default();
2043 tree
2044 });
2045 DbActionPlan::Call {
2046 call: call.clone(),
2047 read,
2048 }
2049 }
2050 DbActionPlan::MutateRead { .. } => return Ok(None),
2051 };
2052
2053 let query = postrust_core::query::build_query(&ActionPlan::Db(unpaged), Some(role))?;
2054 Ok(query.has_main().then(|| query.build_main().0))
2055}
2056
2057/// The total the `Prefer: count` asked for, or `None` when it asked for none.
2058///
2059/// `exact` counts the rows the filters match. `planned` asks the query planner
2060/// what it expects without running anything, which is cheap on a large table
2061/// and correspondingly rough. `estimated` is the planner's guess only where it
2062/// matters: below the server's row ceiling the exact count is already paid for,
2063/// so it is used, and above it the larger of the two is reported -- an estimate
2064/// that came in under a count we have actually taken would be plainly wrong.
2065async fn resolve_count(
2066 tx: &mut sqlx::Transaction<'_, sqlx::Postgres>,
2067 preference: &postrust_core::api_request::PreferCount,
2068 count_sql: &str,
2069 params: &[postrust_sql::SqlParam],
2070 max_rows: Option<i64>,
2071) -> Result<Option<i64>, postrust_core::Error> {
2072 use postrust_core::api_request::PreferCount;
2073
2074 match preference {
2075 PreferCount::Exact => exact_count(tx, count_sql, params).await,
2076 PreferCount::Planned => planned_count(tx, count_sql, params).await,
2077 PreferCount::Estimated => match (exact_count(tx, count_sql, params).await?, max_rows) {
2078 (Some(exact), Some(ceiling)) if exact <= ceiling => Ok(Some(exact)),
2079 (Some(exact), _) => {
2080 let planned = planned_count(tx, count_sql, params).await?;
2081 Ok(Some(exact.max(planned.unwrap_or(exact))))
2082 }
2083 (None, _) => planned_count(tx, count_sql, params).await,
2084 },
2085 }
2086}
2087
2088/// The number of rows the filters match.
2089async fn exact_count(
2090 tx: &mut sqlx::Transaction<'_, sqlx::Postgres>,
2091 count_sql: &str,
2092 params: &[postrust_sql::SqlParam],
2093) -> Result<Option<i64>, postrust_core::Error> {
2094 use sqlx::Row;
2095
2096 let sql = format!("SELECT count(*) FROM ({}) AS pgrst_count", count_sql);
2097 let row = bind_params(sqlx::query(&sql), params)
2098 .fetch_one(&mut **tx)
2099 .await
2100 .map_err(map_sqlx_error)?;
2101 Ok(row.try_get::<i64, _>(0).ok())
2102}
2103
2104/// The number of rows the query planner expects, without running the query.
2105async fn planned_count(
2106 tx: &mut sqlx::Transaction<'_, sqlx::Postgres>,
2107 count_sql: &str,
2108 params: &[postrust_sql::SqlParam],
2109) -> Result<Option<i64>, postrust_core::Error> {
2110 use sqlx::Row;
2111
2112 let sql = format!("EXPLAIN (FORMAT JSON) {}", count_sql);
2113 let row = bind_params(sqlx::query(&sql), params)
2114 .fetch_one(&mut **tx)
2115 .await
2116 .map_err(map_sqlx_error)?;
2117 let plan: serde_json::Value = row.try_get(0).unwrap_or(serde_json::Value::Null);
2118 Ok(plan
2119 .get(0)
2120 .and_then(|node| node.get("Plan"))
2121 .and_then(|node| node.get("Plan Rows"))
2122 .and_then(serde_json::Value::as_i64))
2123}
2124
2125/// Whether this plan changes data.
2126fn is_mutation(db_plan: &postrust_core::plan::DbActionPlan) -> bool {
2127 matches!(
2128 db_plan,
2129 postrust_core::plan::DbActionPlan::MutateRead { .. }
2130 )
2131}
2132
2133/// Whether the caller asked for the affected rows back.
2134fn wants_representation(api_request: &ApiRequest) -> bool {
2135 matches!(
2136 api_request.preferences.representation,
2137 postrust_core::api_request::PreferRepresentation::Full
2138 )
2139}
2140
2141/// Whether a created row's address is worth reporting.
2142///
2143/// `Prefer: return=headers-only` asks for exactly that and nothing else, and
2144/// is the only request PostgREST answers with a `Location`. A caller taking
2145/// the row back reads the key out of the body; a caller wanting neither asked
2146/// for a minimal response and gets one.
2147fn wants_location(api_request: &ApiRequest) -> bool {
2148 matches!(
2149 api_request.preferences.representation,
2150 postrust_core::api_request::PreferRepresentation::HeadersOnly
2151 )
2152}
2153
2154/// The status PostgREST gives a successful mutation, or `None` for anything
2155/// that isn't one.
2156///
2157/// An insert reports 201 Created. An update or delete reports 204 No Content
2158/// unless the caller asked for the rows back, in which case there is content
2159/// to report and it is a plain 200.
2160fn mutation_status(
2161 db_plan: &postrust_core::plan::DbActionPlan,
2162 api_request: &ApiRequest,
2163 created_a_row: bool,
2164 // Whether the statement was in a position to say. A write with no conflict
2165 // clause could only have created rows, and one on a relation that will not
2166 // give up `xmax` cannot tell either way -- in both cases there is nothing
2167 // to report but the creation.
2168 could_have_merged: bool,
2169) -> Option<StatusCode> {
2170 use postrust_core::api_request::PreferResolution;
2171 use postrust_core::plan::{DbActionPlan, MutatePlan};
2172
2173 let DbActionPlan::MutateRead { mutate, .. } = db_plan else {
2174 return None;
2175 };
2176
2177 let merging = matches!(
2178 api_request.preferences.resolution,
2179 Some(PreferResolution::MergeDuplicates)
2180 );
2181
2182 Some(match mutate {
2183 // A `PUT` names the row it is writing, so from the client's side
2184 // nothing was discovered by the response -- it reports the outcome
2185 // only where it asked to see the row.
2186 MutatePlan::Insert { .. } if is_upsert(api_request) => {
2187 match (
2188 wants_representation(api_request),
2189 created_a_row || !could_have_merged,
2190 ) {
2191 (false, _) => StatusCode::NO_CONTENT,
2192 (true, true) => StatusCode::CREATED,
2193 (true, false) => StatusCode::OK,
2194 }
2195 }
2196 // A `POST` that merged into every row it touched created nothing, and
2197 // 201 would be a promise that something new is at the `Location`.
2198 MutatePlan::Insert { .. } if merging && could_have_merged && !created_a_row => {
2199 StatusCode::OK
2200 }
2201 MutatePlan::Insert { .. } => StatusCode::CREATED,
2202 _ => match wants_representation(api_request) {
2203 true => StatusCode::OK,
2204 false => StatusCode::NO_CONTENT,
2205 },
2206 })
2207}
2208
2209/// Ensure every embedded relation's join column is selected on the parent.
2210///
2211/// Returns the columns that were added purely to make the join possible, so
2212/// they can be removed from the response afterwards. An empty select list means
2213/// "all columns", in which case nothing needs adding.
2214fn add_embed_join_columns(
2215 api_request: &mut postrust_core::api_request::ApiRequest,
2216 schema_cache: &postrust_core::SchemaCache,
2217) -> Result<Vec<String>, postrust_core::Error> {
2218 use postrust_core::api_request::{Field, SelectItem};
2219
2220 let Some(parent_qi) = read_target(api_request, None, schema_cache) else {
2221 return Ok(Vec::new());
2222 };
2223
2224 let select = &api_request.query_params.select;
2225 if select.is_empty() {
2226 return Ok(Vec::new());
2227 }
2228
2229 // `*` already names every column, so no join column needs adding -- and
2230 // adding one anyway selects it twice, which makes `src.*` ambiguous. The
2231 // relations are still walked: a computed relationship needs the parent's
2232 // row carried out, and `*` does not provide that.
2233 let selects_everything = select
2234 .iter()
2235 .any(|item| matches!(item, SelectItem::Field { field, .. } if field.name == "*"));
2236
2237 // Only a column selected under its own name serves as a join key. An
2238 // aliased one -- `myId:id` -- arrives in the result as `myId`, so the
2239 // correlation would look for a column the inner query no longer has; the
2240 // same goes for one that was cast or reached into with a JSON path.
2241 let selected: std::collections::HashSet<String> = select
2242 .iter()
2243 .filter_map(|item| match item {
2244 SelectItem::Field {
2245 field,
2246 aggregate: None,
2247 cast: None,
2248 alias: None,
2249 ..
2250 } if field.json_path.is_empty() => Some(field.name.clone()),
2251 _ => None,
2252 })
2253 .collect();
2254
2255 let mut added = Vec::new();
2256
2257 for item in select.clone() {
2258 // A spread joins on a parent column exactly as a plain embed does, so
2259 // it needs that column selected just the same. Without it the parent
2260 // row has no key to match children against and every spread column
2261 // comes back null.
2262 let (relation, hint) = match &item {
2263 SelectItem::Relation { relation, hint, .. } => (relation, hint),
2264 SelectItem::SpreadRelation { relation, hint, .. } => (relation, hint),
2265 SelectItem::Field { .. } => continue,
2266 };
2267
2268 let rel = schema_cache
2269 .find_relationship(&parent_qi, relation, hint.as_deref(), &parent_qi.schema)?
2270 .ok_or_else(|| {
2271 schema_cache.relationship_not_found(
2272 &parent_qi,
2273 relation,
2274 hint.as_deref(),
2275 &parent_qi.schema,
2276 )
2277 })?;
2278
2279 let plan = postrust_core::embed::EmbedPlan::resolve(rel, schema_cache)?;
2280
2281 // A computed relationship joins on nothing -- the parent row is the
2282 // function's argument. That row has to be carried out of the inner
2283 // query, which is a column of its own rather than a join key. After a
2284 // mutation it is carried out of the statement itself, by `RETURNING`,
2285 // because that is the last place the row still has the table's type.
2286 if plan.function.is_some() {
2287 if !added.iter().any(|c| c == PARENT_ROW_COLUMN) {
2288 added.push(PARENT_ROW_COLUMN.to_string());
2289 }
2290 continue;
2291 }
2292
2293 // Every column the join is on, not just the first: a foreign key over
2294 // several columns correlates on all of them, and one left unselected
2295 // is a column the correlation cannot see.
2296 let join_columns: Vec<String> = match plan.columns.is_empty() {
2297 true => vec![plan.local_column.clone()],
2298 false => plan
2299 .columns
2300 .iter()
2301 .map(|(local, _)| local.clone())
2302 .collect(),
2303 };
2304
2305 if selects_everything {
2306 continue;
2307 }
2308
2309 for column in join_columns {
2310 if !selected.contains(&column) && !added.contains(&column) {
2311 api_request.query_params.select.push(SelectItem::Field {
2312 field: Field::simple(&column),
2313 aggregate: None,
2314 aggregate_cast: None,
2315 cast: None,
2316 alias: None,
2317 });
2318 added.push(column);
2319 }
2320 }
2321 }
2322
2323 Ok(added)
2324}
2325
2326/// Build the SELECT-list expressions that embed relations in one query.
2327///
2328/// Returns one `(field_name, expression)` per requested relation. Each
2329/// expression is a correlated subselect yielding JSON, so the whole tree comes
2330/// back from the parent query rather than from a query per relation per level.
2331///
2332/// `alias_counter` hands out a distinct alias per level, so a self-referential
2333/// relationship stays unambiguous.
2334/// Filters addressed at embedded resources, and the parameters they bind.
2335///
2336/// The embed expressions are assembled as SQL text and wrapped around a main
2337/// query that already owns `$1..$n`, so any placeholder introduced here has to
2338/// be renumbered past that point and its value appended in the same order.
2339struct EmbedFilters<'a> {
2340 /// Every path-scoped filter on the request, e.g. `(["clients"], id=eq.1)`.
2341 filters: &'a [(
2342 postrust_core::api_request::EmbedPath,
2343 postrust_core::api_request::Filter,
2344 )],
2345 /// Values bound by the predicates built so far, in placeholder order.
2346 params: Vec<postrust_sql::SqlParam>,
2347 /// How many parameters the main query already uses.
2348 base: usize,
2349 /// Ordering asked of each embedded resource, by path.
2350 orders: &'a [(
2351 postrust_core::api_request::EmbedPath,
2352 Vec<postrust_core::api_request::OrderTerm>,
2353 )],
2354 /// Ranges asked of each embedded resource, by dotted path.
2355 ranges: &'a std::collections::HashMap<String, postrust_core::api_request::Range>,
2356 /// `and=`/`or=` groups asked of each embedded resource, by path.
2357 logic: &'a [(
2358 postrust_core::api_request::EmbedPath,
2359 postrust_core::api_request::LogicTree,
2360 )],
2361 /// Row cap applied to each embedded resource.
2362 max_rows: Option<i64>,
2363 /// Source of unique subquery aliases across the whole embed tree.
2364 alias_counter: usize,
2365}
2366
2367/// Renumber the placeholders in `sql` so they start after `offset`.
2368///
2369/// Scans rather than string-replaces: a naive replacement of `$1` would also
2370/// corrupt the `$1` inside `$10`.
2371fn shift_placeholders(sql: &str, offset: usize) -> String {
2372 let mut out = String::with_capacity(sql.len());
2373 let mut chars = sql.char_indices().peekable();
2374
2375 while let Some((i, ch)) = chars.next() {
2376 if ch != '$' {
2377 out.push(ch);
2378 continue;
2379 }
2380 let start = i + 1;
2381 let mut end = start;
2382 while let Some((j, d)) = chars.peek() {
2383 if d.is_ascii_digit() {
2384 end = j + d.len_utf8();
2385 chars.next();
2386 } else {
2387 break;
2388 }
2389 }
2390 match sql[start..end].parse::<usize>() {
2391 Ok(n) => out.push_str(&format!("${}", n + offset)),
2392 Err(_) => out.push('$'),
2393 }
2394 }
2395
2396 out
2397}
2398
2399impl EmbedFilters<'_> {
2400 /// A fresh alias, unique across the embed tree.
2401 fn next_alias(&mut self) -> String {
2402 self.alias_counter += 1;
2403 format!("e{}", self.alias_counter)
2404 }
2405
2406 /// The `WHERE` fragment for one embedded resource, or `None` if unfiltered.
2407 /// The row window an embedded resource was asked for.
2408 ///
2409 /// The server's own cap still applies, so an embed cannot be asked for
2410 /// more rows than the server is willing to return.
2411 fn range_for(&self, path: &[String], names: &[String]) -> (Option<i64>, i64) {
2412 let range = self
2413 .ranges
2414 .get(&path.join("."))
2415 .or_else(|| self.ranges.get(&names.join(".")));
2416 let requested = range.and_then(|r| r.limit);
2417 let offset = range.map(|r| r.offset).unwrap_or(0);
2418 let limit = match (requested, self.max_rows) {
2419 (Some(limit), Some(cap)) => Some(limit.min(cap)),
2420 (Some(limit), None) => Some(limit),
2421 (None, cap) => cap,
2422 };
2423 (limit, offset)
2424 }
2425
2426 fn predicate_for(
2427 &mut self,
2428 path: &[String],
2429 names: &[String],
2430 child_qi: &postrust_core::api_request::QualifiedIdentifier,
2431 schema_cache: &postrust_core::SchemaCache,
2432 ) -> Result<Option<String>, postrust_core::Error> {
2433 let addresses = |candidate: &[String]| candidate == path || candidate == names;
2434
2435 let matching: Vec<_> = self
2436 .filters
2437 .iter()
2438 .filter(|(filter_path, _)| addresses(filter_path))
2439 .map(|(_, filter)| filter.clone())
2440 .collect();
2441
2442 let matching_logic: Vec<_> = self
2443 .logic
2444 .iter()
2445 .filter(|(logic_path, _)| addresses(logic_path))
2446 .map(|(_, tree)| tree.clone())
2447 .collect();
2448
2449 if matching.is_empty() && matching_logic.is_empty() {
2450 return Ok(None);
2451 }
2452
2453 let table = schema_cache.get_table(child_qi);
2454 let mut parts = Vec::with_capacity(matching.len());
2455
2456 for filter in &matching {
2457 // Inside the child's subselect an unqualified name binds to the
2458 // child, but only because the child has that column: an unknown
2459 // name would resolve outward to the correlated parent and filter
2460 // the wrong table silently. Refuse instead.
2461 let column = table
2462 .and_then(|t| t.columns.get(&filter.field.name))
2463 .ok_or_else(|| {
2464 postrust_core::Error::ColumnNotFound(format!(
2465 "{}.{}",
2466 child_qi.name, filter.field.name
2467 ))
2468 })?;
2469
2470 let frag =
2471 postrust_core::query::QueryBuilder::filter_sql(filter, &column.nominal_type)?;
2472 parts.push(shift_placeholders(
2473 frag.sql(),
2474 self.base + self.params.len(),
2475 ));
2476 self.params.extend(frag.params().iter().cloned());
2477 }
2478
2479 for tree in matching_logic {
2480 // A name the child does not have would bind outward to the
2481 // correlated parent and filter the wrong table, exactly as a plain
2482 // filter would, so the tree is checked before it is rendered.
2483 let mut unknown = None;
2484 check_logic_columns(&tree, table, &mut unknown);
2485 if let Some(name) = unknown {
2486 return Err(postrust_core::Error::ColumnNotFound(format!(
2487 "{}.{}",
2488 child_qi.name, name
2489 )));
2490 }
2491
2492 let resolver = |name: &str| -> String {
2493 table
2494 .and_then(|t| t.get_column(name))
2495 .map(|c| c.nominal_type.clone())
2496 .unwrap_or_else(|| "text".to_string())
2497 };
2498 let frag = postrust_core::query::QueryBuilder::logic_sql(&tree, resolver)?;
2499 parts.push(shift_placeholders(
2500 frag.sql(),
2501 self.base + self.params.len(),
2502 ));
2503 self.params.extend(frag.params().iter().cloned());
2504 }
2505
2506 Ok(Some(format!("({})", parts.join(" AND "))))
2507 }
2508}
2509
2510/// Record the first column a logic tree names that the table does not have.
2511fn check_logic_columns(
2512 tree: &postrust_core::api_request::LogicTree,
2513 table: Option<&postrust_core::schema_cache::Table>,
2514 unknown: &mut Option<String>,
2515) {
2516 use postrust_core::api_request::LogicTree;
2517
2518 match tree {
2519 LogicTree::Expr { children, .. } => {
2520 for child in children {
2521 check_logic_columns(child, table, unknown);
2522 }
2523 }
2524 LogicTree::Stmt(filter) => {
2525 if unknown.is_none()
2526 && table
2527 .map(|t| t.get_column(&filter.field.name).is_none())
2528 .unwrap_or(true)
2529 {
2530 *unknown = Some(filter.field.name.clone());
2531 }
2532 }
2533 }
2534}
2535
2536/// One level of the embed tree.
2537#[derive(Default)]
2538struct EmbedLevel {
2539 /// `(response key, SQL expression)` for each relation at this level.
2540 expressions: Vec<(String, String)>,
2541 /// `EXISTS` predicates from `!inner` on these relations. They reference
2542 /// the alias of the level *above*, so the caller applies them.
2543 inner_joins: Vec<String>,
2544 /// `ORDER BY` expressions for terms naming an embedded resource.
2545 orders: Vec<String>,
2546 /// Each embed under the name the client used for it, with its
2547 /// expression. A spread's response key is internal, so this is what a
2548 /// filter naming the embed -- `?clients=is.null` -- is matched against.
2549 filterable: Vec<(String, String)>,
2550 /// For each spread, the keys it contributes to its parent.
2551 ///
2552 /// Needed because a spread that matched nothing still carries them --
2553 /// as nulls, or as empty arrays for a to-many -- and by then there is no
2554 /// row to read the names off.
2555 spread_columns: Vec<(String, Vec<String>)>,
2556 /// Whether this level had any relation to embed at all.
2557 ///
2558 /// Distinct from having produced an expression: `clients()` is a relation
2559 /// that deliberately contributes nothing, and the fallback path must not
2560 /// take an absent expression for a relation the single-query form could
2561 /// not handle.
2562 saw_relations: bool,
2563}
2564
2565/// The keys a spread's selection contributes to the object above it.
2566///
2567/// `*` names every column of the related table, an alias renames the column it
2568/// precedes, and a nested spread contributes its own keys, having already been
2569/// flattened into this one.
2570fn spread_output_names(
2571 schema_cache: &postrust_core::SchemaCache,
2572 child_qi: &postrust_core::api_request::QualifiedIdentifier,
2573 select: &[postrust_core::api_request::SelectItem],
2574) -> Vec<String> {
2575 use postrust_core::api_request::SelectItem;
2576
2577 let mut names = Vec::new();
2578 for item in select {
2579 match item {
2580 SelectItem::Field { field, .. } if field.name == "*" => {
2581 if let Some(table) = schema_cache.get_table(child_qi) {
2582 names.extend(table.columns.keys().cloned());
2583 }
2584 }
2585 SelectItem::Field { field, alias, .. } => {
2586 names.push(alias.clone().unwrap_or_else(|| field.name.clone()));
2587 }
2588 SelectItem::Relation {
2589 relation, alias, ..
2590 } => names.push(alias.clone().unwrap_or_else(|| relation.clone())),
2591 SelectItem::SpreadRelation {
2592 relation, select, ..
2593 } => {
2594 if let Some(rel) = schema_cache
2595 .find_relationship(child_qi, relation, None, &child_qi.schema)
2596 .ok()
2597 .flatten()
2598 {
2599 let target = rel.foreign_table().clone();
2600 names.extend(spread_output_names(schema_cache, &target, select));
2601 }
2602 }
2603 }
2604 }
2605 names
2606}
2607
2608/// Dissolve spread embeds into the objects that contain them.
2609///
2610/// A spread arrives as an ordinary JSON column under a reserved name. This
2611/// walks the response and, wherever it finds one, moves its columns up into
2612/// the surrounding object and drops the name.
2613///
2614/// It recurses first, so a spread nested inside another is already flattened
2615/// by the time its parent is -- which is what carries a grandchild's columns
2616/// all the way up.
2617///
2618/// Spreading a to-many gives each column an array of that column's values
2619/// across the matched rows, rather than an array of objects.
2620fn flatten_spreads(
2621 value: &mut serde_json::Value,
2622 columns: &std::collections::HashMap<String, Vec<String>>,
2623) {
2624 match value {
2625 serde_json::Value::Array(items) => {
2626 for item in items {
2627 flatten_spreads(item, columns);
2628 }
2629 }
2630 serde_json::Value::Object(object) => {
2631 for (_, child) in object.iter_mut() {
2632 flatten_spreads(child, columns);
2633 }
2634
2635 let spread_keys: Vec<String> = object
2636 .keys()
2637 .filter(|key| key.starts_with(SPREAD_KEY_PREFIX))
2638 .cloned()
2639 .collect();
2640
2641 for key in spread_keys {
2642 let Some(spread) = object.remove(&key) else {
2643 continue;
2644 };
2645 let names = columns.get(&key).cloned().unwrap_or_default();
2646 match spread {
2647 serde_json::Value::Object(columns) => {
2648 for (name, column) in columns {
2649 object.insert(name, column);
2650 }
2651 }
2652 // A to-many: one array per column rather than an array of
2653 // objects, so the rows are transposed. With no rows the
2654 // columns are still named, each holding an empty array.
2655 serde_json::Value::Array(rows) => {
2656 for name in names {
2657 let column = rows
2658 .iter()
2659 .map(|row| {
2660 row.get(&name).cloned().unwrap_or(serde_json::Value::Null)
2661 })
2662 .collect();
2663 object.insert(name, serde_json::Value::Array(column));
2664 }
2665 }
2666 // The relationship matched nothing. The keys are still
2667 // the client's to expect, so they are there, holding null.
2668 _ => {
2669 for name in names {
2670 object.insert(name, serde_json::Value::Null);
2671 }
2672 }
2673 }
2674 }
2675 }
2676 _ => {}
2677 }
2678}
2679
2680/// The relation an embed names, found by the name the response uses.
2681fn embedded_by_name<'a>(
2682 select: &'a [postrust_core::api_request::SelectItem],
2683 name: &str,
2684) -> Option<(&'a str, Option<&'a str>)> {
2685 use postrust_core::api_request::SelectItem;
2686
2687 select.iter().find_map(|item| match item {
2688 SelectItem::Relation {
2689 relation,
2690 alias,
2691 hint,
2692 ..
2693 } if alias.as_deref().unwrap_or(relation) == name => {
2694 Some((relation.as_str(), hint.as_deref()))
2695 }
2696 // A spread has no alias -- nothing is being named -- so it answers to
2697 // the relation's own name.
2698 SelectItem::SpreadRelation { relation, hint, .. } if relation == name => {
2699 Some((relation.as_str(), hint.as_deref()))
2700 }
2701 _ => None,
2702 })
2703}
2704
2705/// Build the `ORDER BY` expressions for terms naming an embedded resource.
2706///
2707/// `order=clients(name)` orders by a column the parent does not have, so each
2708/// term becomes a scalar subselect fetched per parent row -- the same shape as
2709/// the embed itself, reduced to one column.
2710fn build_embed_orders(
2711 schema_cache: &postrust_core::SchemaCache,
2712 parent_qi: &postrust_core::api_request::QualifiedIdentifier,
2713 order: &[postrust_core::plan::CoercibleOrderTerm],
2714 select: &[postrust_core::api_request::SelectItem],
2715 ctx: &mut EmbedFilters<'_>,
2716) -> Result<Vec<String>, postrust_core::Error> {
2717 let mut orders = Vec::new();
2718
2719 for term in order.iter() {
2720 let Some(relation) = &term.relation else {
2721 continue;
2722 };
2723
2724 // `order=` names a resource this request embedded, by the name the
2725 // response gives it -- so `pros:projects(*)` is ordered on as `pros`,
2726 // and the table it came from is what the relationship is looked up
2727 // by. A name the selection does not carry is not an embedded resource
2728 // here, whether or not the schema has a table by that name.
2729 let embedded = embedded_by_name(select, relation)
2730 .ok_or_else(|| postrust_core::Error::NotAnEmbeddedResource(relation.clone()))?;
2731
2732 let rel = schema_cache
2733 .find_relationship(parent_qi, embedded.0, embedded.1, &parent_qi.schema)?
2734 .ok_or_else(|| {
2735 schema_cache.relationship_not_found(
2736 parent_qi,
2737 embedded.0,
2738 embedded.1,
2739 &parent_qi.schema,
2740 )
2741 })?;
2742
2743 // A resource that yields many rows per parent has no single value to
2744 // order on, so there is nothing the request could mean.
2745 if !rel.is_to_one() {
2746 return Err(postrust_core::Error::RelatedOrderNotPossible {
2747 origin: parent_qi.name.clone(),
2748 relation: relation.clone(),
2749 });
2750 }
2751
2752 let plan = postrust_core::embed::EmbedPlan::resolve(rel, schema_cache)?;
2753
2754 let child_alias = ctx.next_alias();
2755
2756 // The column belongs to the related table, so its type has to be read
2757 // from there. Resolved against the parent it would be unknown, and an
2758 // unknown type is one this process asks the database to render -- for
2759 // a `jsonb` column that would wrap it in `to_jsonb` and then reach
2760 // into the wrapper.
2761 let child_qi = postrust_core::api_request::QualifiedIdentifier::new(
2762 &plan.foreign_schema,
2763 &plan.foreign_table,
2764 );
2765 let mut field = term.field.clone();
2766 field.to_json = !field.json_path.is_empty()
2767 && !schema_cache
2768 .get_table(&child_qi)
2769 .and_then(|table| table.get_column(&field.name))
2770 .is_some_and(|column| matches!(column.data_type.as_str(), "json" | "jsonb"));
2771
2772 let column =
2773 postrust_core::query::QueryBuilder::qualified_column_sql(Some(&child_alias), &field);
2774 let mut expression =
2775 plan.order_expression("src", PARENT_ROW_COLUMN_REF, &child_alias, &column);
2776
2777 match term.direction {
2778 Some(postrust_core::api_request::OrderDirection::Desc) => expression.push_str(" DESC"),
2779 Some(postrust_core::api_request::OrderDirection::Asc) => expression.push_str(" ASC"),
2780 None => {}
2781 }
2782 match term.nulls {
2783 Some(postrust_core::api_request::OrderNulls::First) => {
2784 expression.push_str(" NULLS FIRST")
2785 }
2786 Some(postrust_core::api_request::OrderNulls::Last) => {
2787 expression.push_str(" NULLS LAST")
2788 }
2789 None => {}
2790 }
2791
2792 orders.push(expression);
2793 }
2794
2795 Ok(orders)
2796}
2797
2798/// The relationship an `order=relation(column)` term names, ready to order by.
2799///
2800/// Resolving it is the same work whether the term sits at the top level or
2801/// inside an embed, and it is the same mistake to skip: a relation term
2802/// rendered as a bare column name orders by a column of the wrong table --
2803/// silently, where the parent happens to have a column of that name, and with
2804/// `column "cost" does not exist` where it does not.
2805fn related_order_target(
2806 schema_cache: &postrust_core::SchemaCache,
2807 parent_qi: &postrust_core::api_request::QualifiedIdentifier,
2808 select: &[postrust_core::api_request::SelectItem],
2809 relation: &str,
2810 ctx: &mut EmbedFilters<'_>,
2811) -> Result<
2812 (
2813 postrust_core::embed::EmbedPlan,
2814 String,
2815 postrust_core::api_request::QualifiedIdentifier,
2816 ),
2817 postrust_core::Error,
2818> {
2819 // `order=` names a resource this request embedded, by the name the
2820 // response gives it -- so `pros:projects(*)` is ordered on as `pros`, and
2821 // the table it came from is what the relationship is looked up by. A name
2822 // the selection does not carry is not an embedded resource here, whether
2823 // or not the schema has a table by that name.
2824 let embedded = embedded_by_name(select, relation)
2825 .ok_or_else(|| postrust_core::Error::NotAnEmbeddedResource(relation.to_string()))?;
2826
2827 let rel = schema_cache
2828 .find_relationship(parent_qi, embedded.0, embedded.1, &parent_qi.schema)?
2829 .ok_or_else(|| {
2830 schema_cache.relationship_not_found(
2831 parent_qi,
2832 embedded.0,
2833 embedded.1,
2834 &parent_qi.schema,
2835 )
2836 })?;
2837
2838 // A resource that yields many rows per parent has no single value to order
2839 // on, so there is nothing the request could mean.
2840 if !rel.is_to_one() {
2841 return Err(postrust_core::Error::RelatedOrderNotPossible {
2842 origin: parent_qi.name.clone(),
2843 relation: relation.to_string(),
2844 });
2845 }
2846
2847 let plan = postrust_core::embed::EmbedPlan::resolve(rel, schema_cache)?;
2848 let child_alias = ctx.next_alias();
2849 let child_qi = postrust_core::api_request::QualifiedIdentifier::new(
2850 &plan.foreign_schema,
2851 &plan.foreign_table,
2852 );
2853
2854 Ok((plan, child_alias, child_qi))
2855}
2856
2857/// The `ORDER BY` an embedded resource was asked for, if any.
2858///
2859/// `clients.order=name.desc` orders the rows inside the embed, which is a
2860/// property of the child's own subselect rather than of the parent.
2861/// `tasks.order=projects(id).desc` orders them by a column of a table the
2862/// child embeds in turn -- so it is resolved against the child, and rendered
2863/// as the same correlated subselect the top level uses.
2864#[allow(clippy::too_many_arguments)] // each names one part of where the embed sits
2865fn embed_order_sql(
2866 schema_cache: &postrust_core::SchemaCache,
2867 child_qi: &postrust_core::api_request::QualifiedIdentifier,
2868 child_select: &[postrust_core::api_request::SelectItem],
2869 child_alias: &str,
2870 ctx: &mut EmbedFilters<'_>,
2871 path: &[String],
2872 names: &[String],
2873) -> Result<Option<String>, postrust_core::Error> {
2874 use postrust_core::api_request::{OrderDirection, OrderNulls, OrderTerm};
2875
2876 let terms: Vec<OrderTerm> = ctx
2877 .orders
2878 .iter()
2879 .filter(|(p, _)| p.as_slice() == path || p.as_slice() == names)
2880 .flat_map(|(_, terms)| terms.iter().cloned())
2881 .collect();
2882
2883 if terms.is_empty() {
2884 return Ok(None);
2885 }
2886
2887 let mut rendered = Vec::with_capacity(terms.len());
2888 for term in &terms {
2889 let (field, direction, nulls) = match term {
2890 OrderTerm::Field {
2891 field,
2892 direction,
2893 nulls,
2894 }
2895 | OrderTerm::Relation {
2896 field,
2897 direction,
2898 nulls,
2899 ..
2900 } => (field, direction, nulls),
2901 };
2902
2903 let mut sql = match term {
2904 OrderTerm::Field { .. } => postrust_sql::escape_ident(&field.name),
2905 OrderTerm::Relation { relation, .. } => {
2906 let (plan, grandchild_alias, grandchild_qi) =
2907 related_order_target(schema_cache, child_qi, child_select, relation, ctx)?;
2908
2909 // The column belongs to the related table, so its type is read
2910 // from there rather than from the child.
2911 let pg_type = schema_cache
2912 .get_table(&grandchild_qi)
2913 .and_then(|table| table.get_column(&field.name))
2914 .map(|column| column.data_type.clone())
2915 .unwrap_or_default();
2916 let coercible = postrust_core::plan::CoercibleField::from_field(field, &pg_type);
2917 let column = postrust_core::query::QueryBuilder::qualified_column_sql(
2918 Some(&grandchild_alias),
2919 &coercible,
2920 );
2921
2922 // The child is a real table alias at this level, so its alias
2923 // is also the expression for its whole row.
2924 plan.order_expression(child_alias, child_alias, &grandchild_alias, &column)
2925 }
2926 };
2927
2928 match direction {
2929 Some(OrderDirection::Desc) => sql.push_str(" DESC"),
2930 Some(OrderDirection::Asc) => sql.push_str(" ASC"),
2931 None => {}
2932 }
2933 match nulls {
2934 Some(OrderNulls::First) => sql.push_str(" NULLS FIRST"),
2935 Some(OrderNulls::Last) => sql.push_str(" NULLS LAST"),
2936 None => {}
2937 }
2938
2939 rendered.push(sql);
2940 }
2941
2942 Ok(Some(rendered.join(", ")))
2943}
2944
2945#[allow(clippy::too_many_arguments)] // each names one part of where the embed sits
2946fn build_embed_expressions(
2947 schema_cache: &postrust_core::SchemaCache,
2948 parent_qi: &postrust_core::api_request::QualifiedIdentifier,
2949 parent_alias: &str,
2950 // The SQL expression yielding the parent's whole row, which a computed
2951 // relationship takes as its argument. At a nested level the parent is a
2952 // real table alias and so is the row; at the top level the parent is a
2953 // derived table, whose alias is a `record` rather than the table's own
2954 // composite type, so a column carrying the row is passed instead.
2955 parent_row: &str,
2956 select: &[postrust_core::api_request::SelectItem],
2957 ctx: &mut EmbedFilters<'_>,
2958 path: &[String],
2959 // The same path spelled with each relation's own name rather than the
2960 // alias the request gave it.
2961 names: &[String],
2962) -> Result<EmbedLevel, postrust_core::Error> {
2963 use postrust_core::api_request::SelectItem;
2964
2965 let mut level = EmbedLevel::default();
2966
2967 for item in select {
2968 // A spread is embedded exactly like a plain relation. The two differ
2969 // only in where the child's columns end up, and that is settled once
2970 // the rows are JSON -- so here the spread is marked by the name its
2971 // expression is given, and flattened afterwards.
2972 let (relation, alias, child_select, join_type, hint, is_spread) = match item {
2973 SelectItem::Relation {
2974 relation,
2975 alias,
2976 select,
2977 join_type,
2978 hint,
2979 } => (relation, alias.clone(), select, join_type, hint, false),
2980 SelectItem::SpreadRelation {
2981 relation,
2982 select,
2983 join_type,
2984 hint,
2985 } => (relation, None, select, join_type, hint, true),
2986 SelectItem::Field { .. } => continue,
2987 };
2988
2989 level.saw_relations = true;
2990
2991 let rel = schema_cache
2992 .find_relationship(parent_qi, relation, hint.as_deref(), &parent_qi.schema)?
2993 .ok_or_else(|| {
2994 schema_cache.relationship_not_found(
2995 parent_qi,
2996 relation,
2997 hint.as_deref(),
2998 &parent_qi.schema,
2999 )
3000 })?;
3001 let plan = postrust_core::embed::EmbedPlan::resolve(rel, schema_cache)?;
3002
3003 // No row to pass: the caller said the parent's is not expressible
3004 // here, and a computed relationship is a function of it.
3005 if plan.function.is_some() && parent_row.is_empty() {
3006 continue;
3007 }
3008
3009 let child_alias = ctx.next_alias();
3010 let child_qi = postrust_core::api_request::QualifiedIdentifier::new(
3011 &plan.foreign_schema,
3012 &plan.foreign_table,
3013 );
3014
3015 // Filters are addressed by the name the client used, which may be
3016 // either the alias or the relation itself: `c:clients(*)&c.id=eq.1`
3017 // and `the_tasks:tasks(*)&tasks.id=eq.1` both name the same embed.
3018 let mut child_path = path.to_vec();
3019 child_path.push(alias.clone().unwrap_or_else(|| relation.clone()));
3020 let mut child_names = names.to_vec();
3021 child_names.push(relation.clone());
3022 let child_where = ctx.predicate_for(&child_path, &child_names, &child_qi, schema_cache)?;
3023
3024 // Deeper relations first: they become part of this level's SELECT list.
3025 let child_row = postrust_sql::escape_ident(&child_alias);
3026 let nested = build_embed_expressions(
3027 schema_cache,
3028 &child_qi,
3029 &child_alias,
3030 &child_row,
3031 child_select,
3032 ctx,
3033 &child_path,
3034 &child_names,
3035 )?;
3036
3037 // A nested `!inner` is written against this child's alias, which is
3038 // only in scope inside this child's own subselect -- so it narrows the
3039 // child here rather than travelling up to the top-level wrapper.
3040 let child_where = match (child_where, nested.inner_joins.is_empty()) {
3041 (existing, true) => existing,
3042 (Some(existing), false) => Some(format!(
3043 "{} AND {}",
3044 existing,
3045 nested.inner_joins.join(" AND ")
3046 )),
3047 (None, false) => Some(nested.inner_joins.join(" AND ")),
3048 };
3049
3050 // `!inner` on this relation restricts the parent rows. It is emitted
3051 // against `parent_alias`, so it belongs to the caller's level; reusing
3052 // the same predicate string lets both places share placeholders.
3053 // The child keeps the same alias it has in the embed subselect. The
3054 // two are sibling scopes, never nested, so there is no collision --
3055 // and any nested predicate folded into `child_where` already names
3056 // that alias, so it has to be the one in scope here too.
3057 if matches!(join_type, Some(postrust_core::api_request::JoinType::Inner)) {
3058 level.inner_joins.push(plan.inner_join_predicate(
3059 parent_alias,
3060 parent_row,
3061 &child_alias,
3062 child_where.as_deref(),
3063 ));
3064 }
3065
3066 level.spread_columns.extend(nested.spread_columns);
3067 let nested_expressions = nested.expressions;
3068
3069 let child_table =
3070 schema_cache.get_table(&postrust_core::api_request::QualifiedIdentifier::new(
3071 &plan.foreign_schema,
3072 &plan.foreign_table,
3073 ));
3074
3075 // The child's own columns. Empty means every column, which is what a
3076 // relation with no explicit selection asks for.
3077 let mut parts: Vec<String> = Vec::new();
3078 let mut project_everything = child_select.is_empty();
3079 for nested_item in child_select {
3080 match nested_item {
3081 SelectItem::Field { field, .. } if field.name == "*" => {
3082 project_everything = true;
3083 }
3084 SelectItem::Field { field, alias, .. } => {
3085 let column = postrust_sql::escape_ident(&field.name);
3086 // A schema that declared how one of its domains is written
3087 // on the wire meant it wherever the value appears, and an
3088 // embedded row is the same value in a different place.
3089 let rendered = child_table
3090 .as_ref()
3091 .and_then(|table| table.get_column(&field.name))
3092 .and_then(|c| schema_cache.representation(c.representation_type(), "json"))
3093 .map(|function| format!("{}({})", function, column))
3094 .unwrap_or_else(|| column.clone());
3095 let name = alias.as_deref().unwrap_or(&field.name);
3096 parts.push(format!(
3097 "{} AS {}",
3098 rendered,
3099 postrust_sql::escape_ident(name)
3100 ));
3101 }
3102 // Handled as an expression, not a column.
3103 SelectItem::Relation { .. } | SelectItem::SpreadRelation { .. } => {}
3104 }
3105 }
3106 if project_everything {
3107 parts.clear();
3108 parts.push(format!("{}.*", postrust_sql::escape_ident(&child_alias)));
3109 }
3110 for (field_name, expression) in nested_expressions {
3111 parts.push(format!(
3112 "{} AS {}",
3113 expression,
3114 postrust_sql::escape_ident(&field_name)
3115 ));
3116 }
3117
3118 let inner_select = parts.join(", ");
3119 let child_order = embed_order_sql(
3120 schema_cache,
3121 &child_qi,
3122 child_select,
3123 &child_alias,
3124 ctx,
3125 &child_path,
3126 &child_names,
3127 )?;
3128 let (child_limit, child_offset) = ctx.range_for(&child_path, &child_names);
3129 let expression = plan.embed_expression(
3130 parent_alias,
3131 parent_row,
3132 &child_alias,
3133 &inner_select,
3134 child_limit,
3135 child_offset,
3136 child_where.as_deref(),
3137 child_order.as_deref(),
3138 // Positionally, with the parent row and nothing else: this surface
3139 // has no session to supply and no place to write an argument.
3140 None,
3141 )?;
3142
3143 if is_spread {
3144 let names = spread_output_names(schema_cache, &child_qi, child_select);
3145 if names.is_empty() {
3146 // `...clients()` spreads no columns, so there is nothing to
3147 // fetch and nothing to merge.
3148 continue;
3149 }
3150 level
3151 .spread_columns
3152 .push((format!("{}{}", SPREAD_KEY_PREFIX, child_alias), names));
3153 }
3154
3155 let key = match is_spread {
3156 // A name nothing can collide with, since the object it names is
3157 // dissolved into its parent before anyone sees it.
3158 true => format!("{}{}", SPREAD_KEY_PREFIX, child_alias),
3159 false => alias.clone().unwrap_or_else(|| relation.clone()),
3160 };
3161
3162 // Under the name the client used, whatever the response key is: a
3163 // filter naming the embed is written against that name.
3164 //
3165 // What it records is an existence test, not the expression. Asking
3166 // whether a to-many embed is null would never be true -- it renders as
3167 // `[]`, never as null -- and "did this match anything" is the question
3168 // either way.
3169 level.filterable.push((
3170 alias.clone().unwrap_or_else(|| relation.clone()),
3171 plan.inner_join_predicate(
3172 parent_alias,
3173 parent_row,
3174 &child_alias,
3175 child_where.as_deref(),
3176 ),
3177 ));
3178
3179 // `clients()` asks for none of the related resource's columns. It is
3180 // written to narrow the parent -- with `!inner`, or with a filter on
3181 // the embed -- and the resource itself is not part of the answer, so
3182 // it gets no key at all rather than an empty object. The test is
3183 // recursive: an embed whose only content is such an embed contributes
3184 // nothing either.
3185 if !yields_columns(child_select) {
3186 continue;
3187 }
3188
3189 level.expressions.push((key, expression));
3190 }
3191
3192 Ok(level)
3193}
3194
3195/// The media type a function's own return type declares, where the client
3196/// asked for it.
3197///
3198/// A function returning a domain named `text/plain` returns text, and the
3199/// value is the whole response. It is still only returned that way to a client
3200/// that asked: the same value serialised as JSON is a perfectly good answer to
3201/// a request that did not.
3202fn declared_media_type(
3203 db_plan: &DbActionPlan,
3204 api_request: &postrust_core::api_request::ApiRequest,
3205) -> Option<(String, String)> {
3206 let DbActionPlan::Call { call, .. } = db_plan else {
3207 return None;
3208 };
3209 let declared = call.media_type.as_deref()?;
3210 let base = call.media_base_type.clone().unwrap_or_default();
3211 // A function declaring `*/*` says it will render whatever was asked for,
3212 // so every request is one it answers -- and what it answers with is bytes,
3213 // which on the wire is called `application/octet-stream`.
3214 if declared == "*/*" {
3215 return Some(("application/octet-stream".to_string(), base));
3216 }
3217 api_request
3218 .accept_media_types
3219 .iter()
3220 .any(|accepted| accepted.content_type() == declared)
3221 .then(|| (declared.to_string(), base))
3222}
3223
3224/// Whether a selection produces any keys in the response.
3225///
3226/// A relation contributes only what its own selection does, so `tasks()` and
3227/// `tasks(projects())` are alike empty, while `tasks(name)` is not.
3228fn yields_columns(select: &[postrust_core::api_request::SelectItem]) -> bool {
3229 use postrust_core::api_request::SelectItem;
3230
3231 select.iter().any(|item| match item {
3232 SelectItem::Field { .. } => true,
3233 SelectItem::Relation { select, .. } | SelectItem::SpreadRelation { select, .. } => {
3234 yields_columns(select)
3235 }
3236 })
3237}
3238
3239type EmbedFuture<'f> = std::pin::Pin<
3240 Box<dyn std::future::Future<Output = Result<(), postrust_core::Error>> + Send + 'f>,
3241>;
3242
3243/// Attach the relations requested in `select` onto `rows`.
3244///
3245/// One query per relation per level: the parents' join keys are collected and
3246/// passed as a single array, so embedding across a page of parents does not
3247/// become a query per row. Recurses for nested embeds.
3248fn embed_relations<'f>(
3249 conn: &'f mut sqlx::PgConnection,
3250 schema_cache: &'f postrust_core::SchemaCache,
3251 parent_qi: &'f postrust_core::api_request::QualifiedIdentifier,
3252 select: &'f [postrust_core::api_request::SelectItem],
3253 rows: &'f mut [serde_json::Value],
3254 max_rows: Option<i64>,
3255) -> EmbedFuture<'f> {
3256 use postrust_core::api_request::SelectItem;
3257
3258 Box::pin(async move {
3259 if rows.is_empty() {
3260 return Ok(());
3261 }
3262
3263 for item in select {
3264 // A spread embed is fetched exactly like a plain one -- the two
3265 // differ only in where the child's columns end up, which is
3266 // settled once the rows are in hand.
3267 let (relation, alias, nested, is_spread, hint) = match item {
3268 SelectItem::Relation {
3269 relation,
3270 alias,
3271 select: nested,
3272 hint,
3273 ..
3274 } => (relation, alias.clone(), nested, false, hint),
3275 SelectItem::SpreadRelation {
3276 relation,
3277 select: nested,
3278 hint,
3279 ..
3280 } => (relation, None, nested, true, hint),
3281 SelectItem::Field { .. } => continue,
3282 };
3283
3284 let rel = schema_cache
3285 .find_relationship(parent_qi, relation, hint.as_deref(), &parent_qi.schema)?
3286 .ok_or_else(|| {
3287 schema_cache.relationship_not_found(
3288 parent_qi,
3289 relation,
3290 hint.as_deref(),
3291 &parent_qi.schema,
3292 )
3293 })?;
3294
3295 let plan = postrust_core::embed::EmbedPlan::resolve(rel, schema_cache)?;
3296 let keys = postrust_core::embed::parent_keys(rows, &plan.local_column);
3297
3298 // Which of the child's columns the query needs to return.
3299 //
3300 // A column that was not asked for still costs a heap read, JSON
3301 // serialisation in PostgreSQL, socket bytes and a parse before
3302 // being discarded, so the projection is worked out before the query
3303 // rather than filtered afterwards.
3304 //
3305 // A nested embed joins on a column of the child row, so that column
3306 // has to survive the projection even when the client did not ask
3307 // for it. A spread relation falls back to every column: its shape
3308 // is decided further down and is not a plain projection.
3309 let child_qi = postrust_core::api_request::QualifiedIdentifier::new(
3310 &plan.foreign_schema,
3311 &plan.foreign_table,
3312 );
3313
3314 let mut child_columns: Vec<String> = Vec::new();
3315 let mut project_everything = nested.is_empty();
3316 for nested_item in nested {
3317 match nested_item {
3318 SelectItem::Field { field, .. } => child_columns.push(field.name.clone()),
3319 SelectItem::Relation { relation, .. } => {
3320 match schema_cache
3321 .find_relationship(&child_qi, relation, None, &child_qi.schema)
3322 .ok()
3323 .flatten()
3324 {
3325 Some(nested_rel) => {
3326 let nested_plan = postrust_core::embed::EmbedPlan::resolve(
3327 nested_rel,
3328 schema_cache,
3329 )?;
3330 child_columns.push(nested_plan.local_column);
3331 }
3332 // Leave it to the recursive call to report the
3333 // unknown relation, rather than failing here with
3334 // less context.
3335 None => project_everything = true,
3336 }
3337 }
3338 SelectItem::SpreadRelation { .. } => project_everything = true,
3339 }
3340 }
3341 if project_everything {
3342 child_columns.clear();
3343 }
3344
3345 let mut grouped = if keys.is_empty() {
3346 std::collections::HashMap::new()
3347 } else {
3348 let sql = plan.children_grouped_sql(max_rows, &child_columns)?;
3349
3350 let fetched = sqlx::query(&sql)
3351 .bind(&keys)
3352 .fetch_all(&mut *conn)
3353 .await
3354 .map_err(map_sqlx_error)?;
3355
3356 // The query returns the join key and a JSON array of that
3357 // key's children, so the values are read straight out of the
3358 // columns -- the typed row converter would wrap them under the
3359 // expression name.
3360 use sqlx::Row;
3361 let pairs: Vec<(serde_json::Value, serde_json::Value)> = fetched
3362 .into_iter()
3363 .filter_map(|row| {
3364 Some((
3365 row.try_get::<serde_json::Value, _>(0).ok()?,
3366 row.try_get::<serde_json::Value, _>(1).ok()?,
3367 ))
3368 })
3369 .collect();
3370
3371 postrust_core::embed::group_from_aggregated(pairs)
3372 };
3373
3374 // Deeper embeds, if any were asked for.
3375 //
3376 // A nested embed issues one query for every child row at this
3377 // level, so the rows have to be contiguous for it. They arrive
3378 // grouped, so they are flattened for the recursive call and put
3379 // back afterwards. When nothing deeper was requested this is all
3380 // skipped, which is the common case.
3381 let has_deeper_embed = nested.iter().any(|item| {
3382 matches!(
3383 item,
3384 SelectItem::Relation { .. } | SelectItem::SpreadRelation { .. }
3385 )
3386 });
3387
3388 if has_deeper_embed {
3389 let mut order: Vec<(String, usize)> = Vec::with_capacity(grouped.len());
3390 let mut flat: Vec<serde_json::Value> = Vec::new();
3391 for (key, children) in grouped.drain() {
3392 order.push((key, children.len()));
3393 flat.extend(children);
3394 }
3395
3396 embed_relations(
3397 &mut *conn,
3398 schema_cache,
3399 &child_qi,
3400 nested,
3401 &mut flat,
3402 max_rows,
3403 )
3404 .await?;
3405
3406 let mut rest = flat.into_iter();
3407 for (key, count) in order {
3408 grouped.insert(key, rest.by_ref().take(count).collect());
3409 }
3410 }
3411
3412 // Return only the requested columns of the related resource. An
3413 // empty nested select means every column.
3414 //
3415 // The projection in the query covers the columns; this covers what
3416 // is left over, which is the join column when the client did not
3417 // ask for it.
3418 // A spread picks its own columns out when it merges them, and does
3419 // it by the child's own column names, so pruning here would only
3420 // get in the way.
3421 let requested: Option<std::collections::HashSet<String>> =
3422 if nested.is_empty() || is_spread {
3423 None
3424 } else {
3425 Some(
3426 nested
3427 .iter()
3428 .flat_map(|nested_item| match nested_item {
3429 SelectItem::Field { field, alias, .. } => {
3430 vec![alias.clone().unwrap_or_else(|| field.name.clone())]
3431 }
3432 SelectItem::Relation {
3433 relation, alias, ..
3434 } => vec![alias.clone().unwrap_or_else(|| relation.clone())],
3435 // A spread one level down has already merged
3436 // its columns into these rows, so they are
3437 // part of what was asked for.
3438 SelectItem::SpreadRelation { select, .. } => select
3439 .iter()
3440 .filter_map(|column| match column {
3441 SelectItem::Field { field, .. } => Some(field.name.clone()),
3442 _ => None,
3443 })
3444 .collect(),
3445 })
3446 .collect(),
3447 )
3448 };
3449
3450 if let Some(requested) = requested {
3451 for group in grouped.values_mut() {
3452 for child in group.iter_mut() {
3453 if let Some(object) = child.as_object_mut() {
3454 object.retain(|key, _| requested.contains(key));
3455 }
3456 }
3457 }
3458 }
3459 if is_spread {
3460 // The child rows carry the table's own column names, so a
3461 // column is read by its real name and written under its alias.
3462 let spread_columns: Vec<(String, String)> = nested
3463 .iter()
3464 .filter_map(|nested_item| match nested_item {
3465 SelectItem::Field { field, alias, .. } => Some((
3466 field.name.clone(),
3467 alias.clone().unwrap_or_else(|| field.name.clone()),
3468 )),
3469 _ => None,
3470 })
3471 .collect();
3472 for row in rows.iter_mut() {
3473 postrust_core::embed::spread_into_parent(row, &plan, &grouped, &spread_columns);
3474 }
3475 } else {
3476 let field_name = alias.clone().unwrap_or_else(|| relation.clone());
3477 for row in rows.iter_mut() {
3478 postrust_core::embed::attach_to_parent(row, &field_name, &plan, &grouped);
3479 }
3480 }
3481 }
3482
3483 Ok(())
3484 })
3485}
3486
3487/// Bind SqlParam values to a sqlx query.
3488fn bind_params<'q>(
3489 mut query: sqlx::query::Query<'q, sqlx::Postgres, sqlx::postgres::PgArguments>,
3490 params: &'q [postrust_sql::SqlParam],
3491) -> sqlx::query::Query<'q, sqlx::Postgres, sqlx::postgres::PgArguments> {
3492 use postrust_sql::SqlParam;
3493
3494 for param in params {
3495 query = match param {
3496 SqlParam::Null => query.bind(None::<String>),
3497 SqlParam::Bool(b) => query.bind(b),
3498 SqlParam::Int(n) => query.bind(n),
3499 SqlParam::Float(f) => query.bind(f),
3500 SqlParam::Text(s) => query.bind(s),
3501 SqlParam::Bytes(b) => query.bind(b),
3502 SqlParam::Json(j) => query.bind(j),
3503 SqlParam::Uuid(u) => query.bind(u),
3504 SqlParam::Timestamp(t) => query.bind(t),
3505 SqlParam::Array(arr) => {
3506 // Convert array to Vec<String> for text arrays
3507 let strings: Vec<String> = arr
3508 .iter()
3509 .map(|p| match p {
3510 SqlParam::Text(s) => s.clone(),
3511 SqlParam::Int(n) => n.to_string(),
3512 SqlParam::Bool(b) => b.to_string(),
3513 other => format!("{:?}", other),
3514 })
3515 .collect();
3516 query.bind(strings)
3517 }
3518 };
3519 }
3520
3521 query
3522}
3523
3524/// Map sqlx error to our error type.
3525fn map_sqlx_error(e: sqlx::Error) -> postrust_core::Error {
3526 match e {
3527 sqlx::Error::Database(db_err) => {
3528 // Try to downcast to Postgres-specific error for additional details
3529 let (details, hint) = db_err
3530 .try_downcast_ref::<sqlx::postgres::PgDatabaseError>()
3531 .map(|pg_err| {
3532 (
3533 pg_err.detail().map(String::from),
3534 pg_err.hint().map(String::from),
3535 )
3536 })
3537 .unwrap_or((None, None));
3538
3539 postrust_core::Error::Database(postrust_core::error::DatabaseError {
3540 code: db_err.code().map(|c| c.to_string()).unwrap_or_default(),
3541 message: db_err.message().to_string(),
3542 details,
3543 hint,
3544 constraint: db_err.constraint().map(|s| s.to_string()),
3545 table: db_err.table().map(|s| s.to_string()),
3546 column: None,
3547 })
3548 }
3549 other => postrust_core::Error::Internal(other.to_string()),
3550 }
3551}
3552
3553/// Build an HTTP response from our response type.
3554fn build_response(response: PgrstResponse) -> Response {
3555 let mut builder = Response::builder().status(response.status);
3556
3557 for (key, value) in &response.headers {
3558 builder = builder.header(key, value);
3559 }
3560
3561 builder
3562 .body(Body::from(response.body))
3563 .unwrap_or_else(|_| Response::new(Body::empty()))
3564}
3565
3566/// Build an error response.
3567///
3568/// In production mode (PGRST_DEBUG=false or unset), sensitive error details
3569/// are hidden to prevent information leakage.
3570/// Build the response for a failed request.
3571///
3572/// `verbatim_db_errors` passes a database failure through as PostgreSQL
3573/// reported it -- its SQLSTATE, message, detail and hint -- which is what
3574/// PostgREST does and what a client written against PostgREST branches on.
3575/// It is off unless compatibility mode is on, because those fields describe
3576/// the schema rather than the request: a constraint's name, a column that was
3577/// not selected, the text of a failing query. Everything else is reported the
3578/// same either way.
3579fn error_response(error: postrust_core::Error, verbatim_db_errors: bool) -> Response {
3580 let status = error.status_code();
3581
3582 // Check if debug mode is enabled
3583 let debug_mode = std::env::var("PGRST_DEBUG")
3584 .map(|v| v == "true" || v == "1")
3585 .unwrap_or(false);
3586
3587 // A function that raised `sqlstate 'PGRST'` supplied the whole response:
3588 // its status, its headers and its body. Nothing here is the API layer's to
3589 // decide, including whether to sanitise it -- the schema wrote it.
3590 if let postrust_core::Error::Database(db) = &error {
3591 match db.raised_response() {
3592 Some(Ok(raised)) => {
3593 let mut builder = Response::builder()
3594 .status(status)
3595 .header("content-type", "application/json");
3596 for (name, value) in &raised.headers {
3597 builder = builder.header(name, value);
3598 }
3599 return builder
3600 .body(Body::from(
3601 serde_json::to_vec(&raised.body).unwrap_or_default(),
3602 ))
3603 .unwrap_or_else(|_| Response::new(Body::empty()));
3604 }
3605 // The schema meant to write a response and wrote something else.
3606 // Reported as the fault it is, and reported the same way whether
3607 // or not database errors are passed through, since this one is
3608 // about the schema's own text rather than about the request.
3609 Some(Err(fault)) => {
3610 return error_response(
3611 postrust_core::Error::RaiseNotUnderstood(fault),
3612 verbatim_db_errors,
3613 )
3614 }
3615 None => {}
3616 }
3617 }
3618
3619 let body = if debug_mode {
3620 // Full error details in debug mode
3621 serde_json::to_vec(&error.to_json()).unwrap_or_default()
3622 } else if let (true, postrust_core::Error::Database(db)) = (verbatim_db_errors, &error) {
3623 serde_json::to_vec(&serde_json::json!({
3624 "code": db.code,
3625 "message": db.message,
3626 "details": db.details,
3627 "hint": db.hint,
3628 }))
3629 .unwrap_or_default()
3630 } else {
3631 // Sanitized error in production
3632 let sanitized = serde_json::json!({
3633 "code": error.code(),
3634 "message": sanitize_error_message(&error),
3635 "details": error.details(),
3636 // A hint says how to correct the request. It is carried by the
3637 // error itself, so it never has to reach for anything the
3638 // sanitised message deliberately left out.
3639 "hint": error.hint(),
3640 });
3641 serde_json::to_vec(&sanitized).unwrap_or_default()
3642 };
3643
3644 let mut builder = Response::builder()
3645 .status(status)
3646 .header("content-type", "application/json");
3647
3648 // A 401 has to say what would satisfy it, or a client has no way to know
3649 // it should be sending a token at all. Where a token *was* sent and was
3650 // not accepted, the challenge says which -- `invalid_token` with the
3651 // reason, so a client can tell "your key is wrong" from "your clock is"
3652 // without parsing the body.
3653 if status == StatusCode::UNAUTHORIZED {
3654 let challenge = match &error {
3655 postrust_core::Error::InvalidJwt(_)
3656 | postrust_core::Error::NoSuitableJwtKey
3657 | postrust_core::Error::JwtClaim(_) => format!(
3658 "Bearer error=\"invalid_token\", error_description=\"{}\"",
3659 error
3660 ),
3661 _ => "Bearer".to_string(),
3662 };
3663 builder = builder.header("www-authenticate", challenge);
3664 }
3665
3666 builder
3667 .body(Body::from(body))
3668 .unwrap_or_else(|_| Response::new(Body::empty()))
3669}
3670
3671/// Sanitize error messages for production.
3672fn sanitize_error_message(error: &postrust_core::Error) -> String {
3673 use postrust_core::Error;
3674
3675 // A schema-cache miss names what the client asked for, and the client
3676 // already knows that -- it is what it sent. Saying so is what makes the
3677 // difference between "Resource not found" and a message the client can act
3678 // on, and it discloses nothing it did not supply. What stays sanitised is
3679 // the database's own account of a failure, which is about the schema
3680 // rather than about the request.
3681 match error {
3682 Error::TableNotFound { name, .. } | Error::NotFound(name) => {
3683 return format!("Could not find the table '{}' in the schema cache", name)
3684 }
3685 Error::FunctionNotFound {
3686 name,
3687 params,
3688 single_param,
3689 ..
3690 } => {
3691 // A body that is one value carries no argument names, so there is
3692 // no argument list to report -- naming the function is the whole
3693 // of what was looked for.
3694 let looked_for =
3695 match postrust_core::error::names_only_one_param(single_param.as_deref()) {
3696 true => name.clone(),
3697 false => postrust_core::error::function_signature(name, params),
3698 };
3699 return format!(
3700 "Could not find the function {} in the schema cache",
3701 looked_for
3702 );
3703 }
3704 Error::UnacceptableSchema { requested, .. } => {
3705 return format!("Invalid schema: {}", requested)
3706 }
3707 // The candidate list is the whole point of the message: it names the
3708 // signatures that could not be told apart.
3709 Error::AmbiguousFunction { .. } | Error::AmbiguousRelationship { .. } => {
3710 // The candidates are the whole point of these messages: they name
3711 // what could not be told apart.
3712 return error.to_string();
3713 }
3714 _ => {}
3715 }
3716
3717 (match error {
3718 Error::ColumnNotFound(_) => "Column not found",
3719 // The column is the client's own word and the relation is the one it
3720 // addressed, so naming both says nothing it did not send.
3721 Error::UnknownColumn { .. } => return error.to_string(),
3722 // Each of these says only what the request itself said: the resource
3723 // it named, the preference it sent, the shape it asked for. Repeating
3724 // that back tells the client nothing it did not already know, and is
3725 // the difference between a message it can act on and one it cannot.
3726 Error::RelationshipNotFound { .. }
3727 | Error::NotAnEmbeddedResource(_)
3728 | Error::RelatedOrderNotPossible { .. }
3729 | Error::InvalidPreferences(_)
3730 | Error::NotSingular { .. }
3731 | Error::InvalidRange(_)
3732 | Error::InvalidBody(_)
3733 | Error::UnsupportedMethod(_)
3734 | Error::InvalidRpcMethod(_)
3735 | Error::InvalidPutFilters
3736 | Error::PutLimitNotAllowed
3737 | Error::PutMatchingPk
3738 | Error::MaxAffectedExceeded(_)
3739 | Error::InvalidGucHeaders
3740 | Error::InvalidGucStatus
3741 | Error::InvalidMediaType(_)
3742 // Says only that the rows the request asked for carry no geometry,
3743 // which is about the request's own shape and is what the client has
3744 // to change.
3745 | Error::MissingGeometryColumn
3746 // Quotes the client's own URL back at it, and names the grammar the
3747 // API publishes. Neither is the schema's.
3748 | Error::UnparsableQuery { .. }
3749 // Says nothing about the schema beyond that one RAISE is malformed,
3750 // which is what the author has to fix.
3751 | Error::RaiseNotUnderstood(_)
3752 | Error::InvalidResourcePath => return error.to_string(),
3753 Error::InvalidPath(_) => "Invalid request path",
3754 // What the token failed on is the client's own token, so naming it
3755 // costs nothing and is the only way it can tell a bad signature from
3756 // an expired one.
3757 Error::InvalidJwt(_)
3758 | Error::NoSuitableJwtKey
3759 | Error::JwtClaim(_)
3760 | Error::MissingAuth => return error.to_string(),
3761 Error::InsufficientPermissions(_) => "Forbidden",
3762
3763 // A fixed policy message that reveals nothing about the request or the
3764 // schema, so it is passed through verbatim -- and it is what PostgREST
3765 // says, which is what a client branching on it will be matching.
3766 Error::AggregatesNotAllowed => "Use of aggregate functions is not allowed",
3767 Error::InvalidHeader(_) | Error::InvalidQueryParam(_) => "Invalid request",
3768 Error::Database(_) => "Database error",
3769 Error::ConnectionPool(_) => "Service temporarily unavailable",
3770 Error::Internal(_) => "Internal server error",
3771 _ => "An error occurred",
3772 })
3773 .to_string()
3774}
3775
3776#[cfg(test)]
3777mod tests {
3778 /// The REST surface is mounted at `/api`, and an `OPTIONS` reaches the
3779 /// layer before the mount is stripped off.
3780 #[test]
3781 fn the_api_path_is_the_one_below_the_mount() {
3782 assert_eq!(super::api_path("/api/items", false), Some("/items"));
3783 assert_eq!(super::api_path("/api/rpc/f", false), Some("/rpc/f"));
3784 assert_eq!(super::api_path("/api", false), Some("/"));
3785 assert_eq!(super::api_path("/api/", false), Some("/"));
3786 }
3787
3788 /// Answering for a route that is not the API's turns a working preflight
3789 /// into a 404 for a table nobody asked about.
3790 #[test]
3791 fn a_route_that_is_not_the_api_is_left_alone() {
3792 for path in [
3793 "/admin",
3794 "/admin/swagger",
3795 "/_",
3796 "/_/health",
3797 "/api/graphql",
3798 // The endpoints a Hasura deployment has. `/v1/graphql` is the one
3799 // every generated client is pointed at, so answering its preflight
3800 // with "no such table" breaks the surface this branch exists to
3801 // offer -- and only in compatibility mode, where nothing else
3802 // looks wrong.
3803 "/healthz",
3804 "/v1/version",
3805 "/v1/graphql",
3806 "/v1/graphql/ws",
3807 "/v1alpha1/graphql",
3808 ] {
3809 assert_eq!(super::api_path(path, false), None, "{path}");
3810 assert_eq!(super::api_path(path, true), None, "{path}");
3811 }
3812 }
3813
3814 /// The mounts are whole segments here too: a table may be called `v1`.
3815 #[test]
3816 fn a_hasura_mount_does_not_swallow_a_table_beside_it() {
3817 assert_eq!(super::api_path("/v1", true), Some("/v1"));
3818 assert_eq!(super::api_path("/healthzz", true), Some("/healthzz"));
3819 assert_eq!(super::api_path("/v1alpha1", true), Some("/v1alpha1"));
3820 }
3821
3822 /// Outside the mounts the surface exists only in compatibility mode,
3823 /// where PostgREST's own paths answer at the root.
3824 #[test]
3825 fn the_root_is_the_api_only_in_compatibility_mode() {
3826 assert_eq!(super::api_path("/items", true), Some("/items"));
3827 assert_eq!(super::api_path("/", true), Some("/"));
3828 assert_eq!(super::api_path("/items", false), None);
3829 assert_eq!(super::api_path("/", false), None);
3830 }
3831
3832 /// A mount is a path segment: `/apiary` is not under `/api`.
3833 #[test]
3834 fn a_mount_matches_whole_segments_only() {
3835 assert_eq!(super::api_path("/apiary", true), Some("/apiary"));
3836 assert_eq!(
3837 super::api_path("/administrators", true),
3838 Some("/administrators")
3839 );
3840 }
3841
3842 /// A data-modifying `WITH` may only appear at the top level, so anything
3843 /// that wraps the statement has to leave the clause where it is.
3844 #[test]
3845 fn a_leading_cte_is_lifted_off_the_statement() {
3846 let (clause, body) = super::split_leading_cte(
3847 "WITH pgrst_mutation_result AS (DELETE FROM t WHERE n = ')' RETURNING t.id) \
3848 SELECT \"id\" FROM \"pgrst_mutation_result\"",
3849 );
3850
3851 assert_eq!(
3852 clause,
3853 "WITH pgrst_mutation_result AS (DELETE FROM t WHERE n = ')' RETURNING t.id) "
3854 );
3855 assert_eq!(body, "SELECT \"id\" FROM \"pgrst_mutation_result\"");
3856 }
3857
3858 /// A plain read has no clause to lift.
3859 #[test]
3860 fn a_statement_without_a_cte_is_left_alone() {
3861 let sql = "SELECT \"id\" FROM \"test\".\"items\"";
3862 let (clause, body) = super::split_leading_cte(sql);
3863
3864 assert!(clause.is_empty());
3865 assert_eq!(body, sql);
3866 }
3867
3868 use super::*;
3869 use postrust_core::plan::CallParams;
3870 use postrust_core::QualifiedIdentifier;
3871 use serde_json::json;
3872
3873 fn call_plan(name: &str, returns_set: bool) -> CallPlan {
3874 CallPlan {
3875 output_columns: Vec::new(),
3876 media_type: None,
3877 media_base_type: None,
3878 function: QualifiedIdentifier::new("public", name),
3879 params: CallParams::None,
3880 returns_scalar: !returns_set,
3881 return_type: None,
3882 returns_set,
3883 returns_composite: false,
3884 volatility: "Volatile".into(),
3885 param_types: Vec::new(),
3886 returns_void: false,
3887 variadic_params: Vec::new(),
3888 }
3889 }
3890
3891 fn composite_call_plan(name: &str, returns_set: bool) -> CallPlan {
3892 CallPlan {
3893 returns_composite: true,
3894 returns_scalar: false,
3895 return_type: None,
3896 ..call_plan(name, returns_set)
3897 }
3898 }
3899
3900 #[test]
3901 fn unwraps_json_return_to_bare_object() {
3902 // `SELECT * FROM sync(...)` on a json-returning function yields a single
3903 // column named after the function.
3904 let rows = vec![json!({"sync": {"ok": true, "count": 3}})];
3905 let (rows, singular) = unwrap_rpc_rows(rows, &call_plan("sync", false));
3906 assert!(singular, "non-set-returning function should be singular");
3907 assert_eq!(rows, vec![json!({"ok": true, "count": 3})]);
3908 }
3909
3910 #[test]
3911 fn unwraps_scalar_return() {
3912 let rows = vec![json!({"add": 42})];
3913 let (rows, singular) = unwrap_rpc_rows(rows, &call_plan("add", false));
3914 assert!(singular);
3915 assert_eq!(rows, vec![json!(42)]);
3916 }
3917
3918 #[test]
3919 fn unwraps_setof_scalar_to_array() {
3920 let rows = vec![json!({"gen": 1}), json!({"gen": 2})];
3921 let (rows, singular) = unwrap_rpc_rows(rows, &call_plan("gen", true));
3922 assert!(!singular, "set-returning function should not be singular");
3923 assert_eq!(rows, vec![json!(1), json!(2)]);
3924 }
3925
3926 #[test]
3927 fn leaves_multi_column_rows_untouched() {
3928 // `RETURNS TABLE(...)` / composite set: rows already have real columns
3929 // and must not be un-nested.
3930 let rows = vec![json!({"id": 1, "name": "a"}), json!({"id": 2, "name": "b"})];
3931 let (out, singular) =
3932 unwrap_rpc_rows(rows.clone(), &composite_call_plan("list_users", true));
3933 assert!(!singular);
3934 assert_eq!(out, rows);
3935 }
3936
3937 #[test]
3938 fn leaves_table_column_named_like_function_untouched() {
3939 // `CREATE FUNCTION foo() RETURNS TABLE(foo int)`: the single output
3940 // column legitimately shares the function's name. The composite
3941 // return-type metadata must prevent it from being mistaken for the
3942 // function-name wrapper.
3943 let rows = vec![json!({"foo": 1}), json!({"foo": 2})];
3944 let (out, singular) = unwrap_rpc_rows(rows.clone(), &composite_call_plan("foo", true));
3945 assert!(!singular);
3946 assert_eq!(out, rows);
3947 }
3948
3949 #[test]
3950 fn single_composite_return_is_singular_but_not_unwrapped() {
3951 // A non-set function returning a row type expands to its columns;
3952 // nothing to un-nest, but the result still renders as a bare object.
3953 let rows = vec![json!({"id": 1, "name": "a"})];
3954 let (out, singular) =
3955 unwrap_rpc_rows(rows.clone(), &composite_call_plan("get_user", false));
3956 assert!(singular);
3957 assert_eq!(out, rows);
3958 }
3959
3960 #[test]
3961 fn leaves_single_key_row_untouched_when_key_is_not_function_name() {
3962 // A single real column that happens to be the only column should not be
3963 // mistaken for the function-name wrapper.
3964 let rows = vec![json!({"id": 7})];
3965 let (out, _) = unwrap_rpc_rows(rows.clone(), &call_plan("get_thing", false));
3966 assert_eq!(out, rows);
3967 }
3968}