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