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