postrust_engine/

handler.rs

//! Request handling: from an HTTP request to the response the REST API gives
//! it, with no web framework in between.

use crate::state::State;
use bytes::Bytes;
use http::{Response, StatusCode};
use http_body_util::BodyExt;
use postrust_core::db::{Db, Tx};
use postrust_core::{
    create_action_plan, parse_request, ActionPlan, ApiRequest, CallPlan, DbActionPlan,
};
use postrust_response::{format_response, QueryResult, Response as PgrstResponse};
use tracing::{debug, debug_span, error, field::Empty, Instrument};

/// Column carrying the parent's whole row, for computed relationships.
///
/// A computed relationship is a function taking the parent row. By the time an
/// embed expression runs, the parent is a derived table, and a derived table's
/// alias has type `record` -- which PostgreSQL will not cast to the table's
/// composite type. So the row is captured one level down, where the real table
/// is still in scope and a bare reference to it yields the composite, and the
/// embed expression reads it from there. It is stripped from the response like
/// any other column added for embedding.
///
/// Defined in `postrust-core` beside the `RETURNING` that emits it for a
/// mutation, so the two spellings cannot drift.
use postrust_core::query::PARENT_ROW_COLUMN;

/// Marks an embedded object whose columns belong to its parent.
///
/// A spread has no key of its own in the response -- its columns land in the
/// parent object. In SQL it is still one JSON column like any other embed, so
/// it is given a name nothing can collide with and dissolved into its parent
/// once the rows are JSON. Doing it there rather than in SQL is what lets a
/// spread nest inside another, and lets it spread a computed relationship.
const SPREAD_KEY_PREFIX: &str = "pgrst_spread_";
const PARENT_ROW_COLUMN_REF: &str = "\"src\".\"pgrst_parent_row\"";

/// Answer a request to the REST API.
///
/// The path is the API's own -- `/items`, `/rpc/f` -- with whatever prefix the
/// host mounted it under already taken off. The body is read here, after the
/// token has been checked, and no further than `max_body_size`.
pub async fn handle<S, B>(state: &S, request: http::Request<B>) -> Response<Bytes>
where
    S: State,
    B: http_body::Body<Data = Bytes>,
    B::Error: Into<Box<dyn std::error::Error + Send + Sync>>,
{
    let method = request.method().clone();
    let path = request.uri().path().to_string();

    debug!("{} {}", method, path);

    let verbatim_db_errors = state.config().compat_mode;

    // What a 405 needs to say what the resource does take: the address and
    // the schema it was asked in, which are all `allow` reads. Kept before the
    // request is consumed; both are cheap to clone.
    let uri = request.uri().clone();
    let profiles: Vec<_> = ["accept-profile", "content-profile"]
        .into_iter()
        .filter_map(|name| Some((name, request.headers().get(name)?.clone())))
        .collect();

    match process_request(state, request).await {
        Ok(response) => response,
        Err(e) => {
            let mut response = error_response(e, verbatim_db_errors);
            // A 405 must carry `Allow` (RFC 9110, section 15.5.6), and it is
            // the answer an OPTIONS on the same resource gives -- whichever
            // refusal it was: a function called with the wrong method, a PUT
            // with filters it cannot take, a view that cannot be written.
            if response.status() == StatusCode::METHOD_NOT_ALLOWED {
                let mut probe = http::Request::builder()
                    .method(http::Method::OPTIONS)
                    .uri(uri);
                for (name, value) in profiles {
                    probe = probe.header(name, value);
                }
                if let Ok(probe) = probe.body(Bytes::new()) {
                    if let Ok(Ok(allow)) = allow(state, probe)
                        .await
                        .map(|a| http::HeaderValue::from_str(&a))
                    {
                        response.headers_mut().insert(http::header::ALLOW, allow);
                    }
                }
            }
            response
        }
    }
}

/// Process a request and return a response.
async fn process_request<S, B>(
    state: &S,
    request: http::Request<B>,
) -> Result<Response<Bytes>, postrust_core::Error>
where
    S: State,
    B: http_body::Body<Data = Bytes>,
    B::Error: Into<Box<dyn std::error::Error + Send + Sync>>,
{
    // Authenticate, before the body is read. The same function serves a
    // host's own routes, so a token means the same thing to both.
    let auth_result = authenticate(state, request.headers()).await?;
    debug!("Authenticated as role: {}", auth_result.role);

    // Parse request.
    //
    // `max_body_size`, not a constant, and not `DefaultBodyLimit` either: that
    // layer works by setting an extension the body *extractors* read, and this
    // handler takes the whole `Request` and reads the body itself. The limit
    // was hardcoded at 10 MiB here, so `PGRST_MAX_BODY_SIZE` was configuring
    // nothing on this path however it was set.
    let (parts, body) = request.into_parts();
    // Still `InvalidBody`, so still a 400: `to_bytes` reports a body that
    // exceeded the limit and a connection that died mid-body as the same
    // error, and answering 413 to a truncated upload would be a worse lie than
    // the 400.
    //
    // What `axum::body::to_bytes` does, spelled out so it needs no axum: the
    // same limit, the same collection, and the same error text.
    let body_bytes = http_body_util::Limited::new(body, state.config().max_body_size)
        .collect()
        .await
        .map(|collected| collected.to_bytes())
        .map_err(|e| postrust_core::Error::InvalidBody(e.to_string()))?;

    // Build HTTP request for parsing
    let mut builder = http::Request::builder()
        .method(parts.method.clone())
        .uri(parts.uri.clone());

    for (key, value) in &parts.headers {
        builder = builder.header(key, value);
    }

    let http_request = builder
        .body(body_bytes.clone())
        .map_err(|e| postrust_core::Error::Internal(e.to_string()))?;

    // Parse API request
    let mut api_request = parse_request(
        &http_request,
        state.default_schema(),
        state.schemas(),
        state.config().db_max_rows,
    )?;
    // For the database's `request.trace_id`. The host put the context in the
    // extensions; `parse_request` saw a request rebuilt without them.
    api_request.trace_id = parts
        .extensions
        .get::<postrust_core::trace::TraceContext>()
        .map(|context| context.trace_id);

    // Parse payload
    if !body_bytes.is_empty() {
        let payload = postrust_core::api_request::payload::parse_payload(
            body_bytes,
            &api_request.content_media_type,
        )?;
        api_request.payload = payload;
    }

    // A body that names different columns row by row cannot be written by one
    // statement. Skipped when `?columns=` says which columns to write, since
    // then the rows are free to differ.
    if api_request.query_params.columns.is_none() {
        if let Some(payload) = &api_request.payload {
            postrust_core::api_request::payload::validate_uniform_keys(payload)?;
        }
    }

    // A write with nothing to write is refused here rather than left to
    // produce an `UPDATE ... SET` with no assignments, which PostgreSQL
    // reports as a syntax error -- true, and no use at all to the client.
    if api_request.payload.is_none() && needs_payload(&api_request) {
        return Err(postrust_core::Error::InvalidBody(
            "Empty or invalid json".into(),
        ));
    }

    // `application/vnd.pgrst.plan` asks for the query plan instead of the
    // result. It is off unless a server turns it on -- it says a great deal
    // about the schema and the data -- and a server that has not is required
    // to say so rather than answer with something else.
    //
    // Checked after the body, not before it: a request that could not be read
    // at all has not got as far as asking for anything, and PostgREST answers
    // it with what is wrong with the request rather than with what the server
    // declines to serve.
    if let Some(plan) = requested_plan(&api_request) {
        return Err(postrust_core::Error::InvalidMediaType(plan));
    }

    // Get schema cache
    let schema_cache = state.schema_cache().await;

    // Embedding joins on a column of the parent row, so that column has to be
    // selected even when the client did not ask for it. Any column added this
    // way is removed from the response once the embed is attached.
    let added_join_columns = add_embed_join_columns(&mut api_request, &schema_cache)?;

    if !state.config().db_aggregates_enabled && selects_an_aggregate(&api_request) {
        return Err(postrust_core::Error::AggregatesNotAllowed);
    }

    // Create execution plan
    let plan = debug_span!("plan").in_scope(|| create_action_plan(&api_request, &schema_cache))?;

    // Released before the query runs, not held through it. tokio's `RwLock`
    // queues a reader behind any writer already waiting, so a schema reload
    // arriving mid-request used to wait for this guard while `execute_plan`
    // waited behind the reload for a guard of its own: neither moved again,
    // and every request after them queued too. Passing this guard down would
    // end the deadlock but keep the lock for as long as the query runs, and a
    // reload waiting on one slow request would hold up every new one behind
    // it. So the guard goes here, and `execute_plan` takes its own briefly
    // where it needs the cache. A reload landing in between can mean the plan
    // was made from the old cache and is finished against the new one; the
    // schema changed under the request either way, and the database is what
    // says whether the statement still holds.
    drop(schema_cache);

    // Execute plan
    let (result, tx) = execute_plan(
        state,
        &api_request,
        &plan,
        &auth_result,
        &added_join_columns,
    )
    .instrument(db_span(&plan))
    .await?;

    // A read has nothing for a refusal to undo -- it ran read-only -- so its
    // transaction ends now, and the connection goes back before the response
    // is serialised rather than after. A plain read has already committed,
    // with its query; this is a read that had more to do after it: a count,
    // embeds fetched by a second query, a rendered media type.
    let tx = match tx {
        Some(tx) if is_read_only(&api_request) => {
            tx.commit()
                .instrument(commit_span())
                .await
                .map_err(postrust_core::Error::from)?;
            None
        }
        tx => tx,
    };

    // Format response
    // A singular response the result cannot satisfy is the client's business,
    // not an internal failure: it asked for one object and the query answered
    // with some other number of rows.
    let render_span = debug_span!("render");
    let response = render_span.in_scope(|| format_response(&api_request, &result));
    drop(render_span);
    let response = response.map_err(|e| match e {
        postrust_response::FormatError::MultipleRows | postrust_response::FormatError::NotFound => {
            // How many rows there were, not how many were kept: a write the
            // caller wanted no representation from still touched them, and
            // that count is what the client got wrong.
            postrust_core::Error::NotSingular {
                rows: result.rows.len().max(result.affected),
            }
        }
        other => postrust_core::Error::Internal(other.to_string()),
    })?;

    // The transaction commits last, once nothing is left that can refuse the
    // request. A write the client is told failed -- more rows than
    // `max-affected` allowed, several where it asked for one object -- must
    // not have happened, and PostgREST rolls it back. Committed any earlier, it
    // stayed written behind the error.
    //
    // A failure here is the database's answer, not the pool's: a deferred
    // constraint is checked at COMMIT, and a serializable transaction can be
    // refused there too. Each is reported with its own SQLSTATE and status --
    // a deferred foreign key is a 409 like an immediate one. Only a
    // connection that broke is reported as the database being unavailable.
    if let Some(tx) = tx {
        tx.commit()
            .instrument(commit_span())
            .await
            .map_err(postrust_core::Error::from)?;
    }

    Ok(build_response(response))
}

/// The span a `COMMIT` sent on its own runs in.
///
/// A sibling of `db.query` beneath the request, not a child of it: the
/// commit happens after the plan's statements are done -- for a write, after
/// the response is rendered, so that a refusal can still roll it back -- and
/// a `db.query` kept open until then would claim the rendering's time as the
/// database's. A read whose transaction ends with its query commits in the
/// same flight, and has no span of its own for it.
fn commit_span() -> tracing::Span {
    debug_span!(
        "db.commit",
        "db.system" = "postgresql",
        "db.operation.name" = "COMMIT",
    )
}

/// The span the count `Prefer: count` asks for runs in, beneath the plan's
/// `db.query`. Its own span, apart from the main statement's, so a trace
/// shows which of the request's round trips took the time -- an exact count
/// over a large table is often most of it.
///
/// `db.operation.name` says how it was counted: `count` for `exact`,
/// `explain` for `planned`; `estimated` may run both, and opens one span for
/// each.
fn count_span(operation: &'static str) -> tracing::Span {
    debug_span!(
        "db.count",
        "db.system" = "postgresql",
        "db.operation.name" = operation,
        "db.query.text" = Empty,
    )
}

/// The span the query fetching one embedded relation runs in, beneath the
/// plan's `db.query`: one per relation per level, for the embeds the main
/// statement could not carry. `db.collection.name` is the table the
/// relation reads.
fn embed_span(
    relation: &str,
    collection: &postrust_core::api_request::QualifiedIdentifier,
) -> tracing::Span {
    debug_span!(
        "db.embed",
        "db.system" = "postgresql",
        "db.operation.name" = "select",
        "db.collection.name" = %collection,
        "postrust.embed.relation" = relation,
        "db.query.text" = Empty,
    )
}

/// The span a plan's database work runs in, named and described as
/// OpenTelemetry's database conventions have it.
///
/// `db.operation.name` and `db.collection.name` rather than the older
/// `db.operation` and `db.sql.table`: the current conventions, and the names
/// an exporter added later will expect. A plan that never reaches the
/// database -- `OPTIONS`, the OpenAPI document -- gets no span.
///
/// `db.query.text` is declared empty and filled in by `execute_plan` only when
/// `otel_db_statement` asks for it.
fn db_span(plan: &ActionPlan) -> tracing::Span {
    use postrust_core::plan::MutatePlan;

    let (operation, collection) = match plan {
        ActionPlan::Info(_) => return tracing::Span::none(),
        ActionPlan::Db(DbActionPlan::Read(tree)) => ("select", &tree.root.from),
        ActionPlan::Db(DbActionPlan::MutateRead { mutate, .. }) => match mutate {
            MutatePlan::Insert { target, .. } => ("insert", target),
            MutatePlan::Update { target, .. } => ("update", target),
            MutatePlan::Delete { target, .. } => ("delete", target),
        },
        ActionPlan::Db(DbActionPlan::Call { call, .. }) => ("call", &call.function),
    };
    debug_span!(
        "db.query",
        "db.system" = "postgresql",
        "db.operation.name" = operation,
        "db.collection.name" = %collection,
        "db.query.text" = Empty,
    )
}

/// The plan media type a request asked for, when that is all it will accept.
///
/// A request that would also take JSON gets JSON; one that will take only the
/// plan is refused, and the message names what it asked for.
fn requested_plan(api_request: &ApiRequest) -> Option<String> {
    use postrust_core::api_request::MediaType;

    // Named back with their parameters, since that is what was asked for: a
    // plan for CSV and a plan for JSON are different requests, and a client
    // told only that "application/vnd.pgrst.plan+json" is unavailable has not
    // been told which of its requests was refused.
    let accepted: Vec<String> = api_request
        .accept_media_types
        .iter()
        .filter(|media| matches!(media, MediaType::Plan { .. }))
        .map(MediaType::to_mime)
        .collect();

    match accepted.len() == api_request.accept_media_types.len() && !accepted.is_empty() {
        true => Some(accepted.join(", ")),
        false => None,
    }
}

/// Which kind of mutation this request is, if it is one.
fn mutation_kind(api_request: &ApiRequest) -> Option<postrust_core::api_request::Mutation> {
    use postrust_core::api_request::{Action, DbAction};

    match &api_request.action {
        Action::Db(DbAction::RelationMut { mutation, .. }) => Some(mutation.clone()),
        _ => None,
    }
}

/// Split a leading `WITH ... AS (...)` off the front of a statement.
///
/// Returns the clause -- with its trailing space, or empty when there is none
/// -- and the statement that follows it. A `WITH` containing an `INSERT`,
/// `UPDATE` or `DELETE` may only appear at the top level, so anything that
/// wraps the statement has to wrap what follows the clause and put the clause
/// back in front.
fn split_leading_cte(sql: &str) -> (String, &str) {
    let Some(open) = sql.strip_prefix("WITH ").and_then(|rest| rest.find('(')) else {
        return (String::new(), sql);
    };
    let open = open + "WITH ".len();

    let mut depth = 0usize;
    let mut quoted = None::<char>;
    for (index, ch) in sql[open..].char_indices() {
        match (quoted, ch) {
            (Some(quote), ch) if ch == quote => quoted = None,
            (Some(_), _) => {}
            (None, '\'') | (None, '"') => quoted = Some(ch),
            (None, '(') => depth += 1,
            (None, ')') => {
                depth -= 1;
                if depth == 0 {
                    let end = open + index + 1;
                    return (format!("{} ", &sql[..end]), sql[end..].trim_start());
                }
            }
            _ => {}
        }
    }

    (String::new(), sql)
}

/// The prefix given to primary key columns added only to build `Location`.
const LOCATION_KEY_PREFIX: &str = "pgrst_location_";

/// The column a media handler's statement carries its row count in.
///
/// Named like the other bookkeeping columns so it cannot collide with anything
/// the request selected. Never reaches the client: the response body is the
/// rendered value, read from the first column.
const MEDIA_ROW_COUNT_COLUMN: &str = "pgrst_media_rows";

/// Whether this request writes the row its URL names.
fn is_upsert(api_request: &ApiRequest) -> bool {
    use postrust_core::api_request::{Action, DbAction, Mutation};

    matches!(
        &api_request.action,
        Action::Db(DbAction::RelationMut {
            mutation: Mutation::SingleUpsert,
            ..
        })
    )
}

/// The `Location` of a newly created row, and removal of the key columns that
/// were added to find it.
///
/// `/projects?id=eq.7` -- the address the row can be read back from, which is
/// what a 201 is required to give. A key column that came back null is
/// addressed with `is.null`, since `eq.` would match nothing.
fn build_location(
    api_request: &ApiRequest,
    rows: &mut [serde_json::Value],
    keys: &[String],
) -> Option<String> {
    if keys.is_empty() {
        return None;
    }

    let table = api_request.path.rsplit('/').next()?;
    let mut conditions = Vec::with_capacity(keys.len());

    for key in keys {
        let value = rows
            .first()?
            .get(key)
            .or_else(|| rows.first()?.get(format!("{}{}", LOCATION_KEY_PREFIX, key)))?;
        conditions.push(match value {
            serde_json::Value::Null => format!("{}=is.null", urlencode(key)),
            serde_json::Value::String(text) => {
                format!("{}=eq.{}", urlencode(key), urlencode(text))
            }
            other => format!("{}=eq.{}", urlencode(key), urlencode(&other.to_string())),
        });
    }

    for row in rows.iter_mut() {
        if let Some(object) = row.as_object_mut() {
            for key in keys {
                object.remove(&format!("{}{}", LOCATION_KEY_PREFIX, key));
            }
        }
    }

    Some(format!("/{}?{}", table, conditions.join("&")))
}

/// Percent-encode everything a query-string value may not carry literally.
fn urlencode(value: &str) -> String {
    let mut encoded = String::with_capacity(value.len());
    for byte in value.as_bytes() {
        match byte {
            b'A'..=b'Z' | b'a'..=b'z' | b'0'..=b'9' | b'-' | b'_' | b'.' | b'~' => {
                encoded.push(*byte as char)
            }
            other => encoded.push_str(&format!("%{:02X}", other)),
        }
    }
    encoded
}

/// Whether this request must carry a body.
///
/// Insert, update and upsert all write values that can only come from one.
/// A delete names its rows in the query string, and a function call may take
/// no arguments at all.
fn needs_payload(api_request: &ApiRequest) -> bool {
    use postrust_core::api_request::{Action, DbAction, Mutation};

    matches!(
        &api_request.action,
        Action::Db(DbAction::RelationMut {
            mutation: Mutation::Create | Mutation::Update | Mutation::SingleUpsert,
            ..
        })
    )
}

/// The `Authorization` header a request carries, as authentication reads it.
///
/// A value that is not visible ASCII is no header at all: it cannot be a
/// bearer token, and it is answered as a request that sent none.
pub fn authorization(headers: &http::HeaderMap) -> Option<&str> {
    headers
        .get(http::header::AUTHORIZATION)
        .and_then(|v| v.to_str().ok())
}

/// Who a request is, from its `Authorization` header.
///
/// The one place a request is authenticated: REST calls this, and so does a
/// host's own route that wants its caller to be who REST would say it is --
/// the server's `User` extractor. Through the token cache where one is
/// configured, the anonymous role when no token is sent, and a failure
/// classified as PostgREST classifies it, for [`error_response`] to answer.
///
/// Keys fetched from a JWKS URL may need fetching first: the one await in
/// authenticating, and one that awaits nothing for a token whose key is
/// already held. See `postrust_auth::jwks`.
///
/// The `auth` span is at debug, beneath the request's own span at info: with
/// the default filter it is never created, and a disabled span is a check of
/// a cached flag. Its fields are evaluated only when something is listening.
/// `jwt.cache` says whether the token cache answered: `hit`, `miss`, or
/// `bypassed` for a request with no token to look up. Absent where no cache
/// is configured. The cache counts the same outcomes for a metric; see
/// `JwtCache::stats`.
pub async fn authenticate<S: State>(
    state: &S,
    headers: &http::HeaderMap,
) -> Result<postrust_auth::AuthResult, postrust_core::Error> {
    let header = authorization(headers);
    postrust_auth::ensure_keys(header, state.jwt_config()).await;
    let auth_span = debug_span!("auth", role = Empty, "jwt.cache" = Empty);
    let auth_result = auth_span
        .in_scope(|| match state.jwt_cache() {
            Some(cache) => {
                let (result, outcome) = cache.authenticate_with_outcome(header, state.jwt_config());
                auth_span.record("jwt.cache", outcome.as_str());
                result
            }
            None => postrust_auth::authenticate(header, state.jwt_config()),
        })
        .map_err(jwt_error)?;
    auth_span.record("role", auth_result.role.as_str());
    Ok(auth_result)
}

/// Classify a JWT failure the way PostgREST reports it.
///
/// A token that could not be read at all and one that was read and found
/// wanting are different answers to the client: the first says the credential
/// is unusable, the second says what about it was unacceptable. PostgREST
/// gives them different codes, and clients branch on them.
fn jwt_error(error: postrust_auth::JwtError) -> postrust_core::Error {
    use postrust_auth::JwtError;

    match error {
        JwtError::NoIdentity => postrust_core::Error::MissingAuth,
        JwtError::SecretMissing => postrust_core::Error::JwtSecretMissing,
        // Read, and found wanting: the token itself is intact, and what it
        // claims is what was not accepted.
        JwtError::Claim(fault) => postrust_core::Error::JwtClaim(fault.to_string()),
        JwtError::InvalidSignature => {
            postrust_core::Error::InvalidJwt("JWT cryptographic operation failed".into())
        }
        JwtError::InvalidHeaderFormat => {
            postrust_core::Error::InvalidJwt("Unsupported token type".into())
        }
        JwtError::NoSuitableKey => postrust_core::Error::NoSuitableJwtKey,
    }
}

/// Execute an action plan.
///
/// Returns the transaction it ran in, still open, for the caller to commit
/// once the response is known to be one it can send. `None` where there is
/// nothing to commit: no statement ran, or a read already committed.
async fn execute_plan<'s, S: State>(
    state: &'s S,
    api_request: &ApiRequest,
    plan: &ActionPlan,
    auth: &postrust_auth::AuthResult,
    added_join_columns: &[String],
) -> Result<(QueryResult, Option<<S::Db as Db>::Tx<'s>>), postrust_core::Error> {
    match plan {
        ActionPlan::Db(db_plan) => {
            // A media type the schema renders itself replaces the whole
            // response body: the aggregate is applied over the rows the
            // request selected, so it sees exactly what would otherwise have
            // been serialised as JSON.
            let media_handler = {
                let schema_cache = state.schema_cache().await;
                read_target(api_request, Some(db_plan), &schema_cache).and_then(|qi| {
                    api_request.accept_media_types.iter().find_map(|media| {
                        schema_cache
                            .media_handler(
                                &api_request.schema,
                                media.content_type(),
                                &qi,
                                has_default_select(&api_request.query_params.select),
                            )
                            .map(|(handler, resolved)| {
                                (
                                    resolved.to_string(),
                                    format!(
                                        "{}.{}",
                                        postrust_sql::escape_ident(&handler.aggregate.schema),
                                        postrust_sql::escape_ident(&handler.aggregate.name)
                                    ),
                                    // Applied to the table's own row, not to
                                    // the derived one, whenever the derived
                                    // row would be a worse description of it.
                                    // A handler naming a table needs the row
                                    // type by name; one taking `anyelement`
                                    // needs the columns to still *be* what
                                    // they are -- a handler calling
                                    // `ST_AsGeoJSON` cannot find a geometry in
                                    // a row whose geometry this side already
                                    // rendered to jsonb. Only where the
                                    // request did not reshape the rows, since
                                    // then the two are the same rows.
                                    handler.table.is_some()
                                        || has_default_select(&api_request.query_params.select),
                                    handler.base_type.clone(),
                                )
                            })
                    })
                })
            };

            // Content negotiation. A media type is rendered by a builtin, by a
            // handler the schema declared, or by the function's own return
            // type -- and a request that names only types nothing here can
            // render has not been answered by serving it JSON under a
            // `Content-Type` it never asked for. PostgREST refuses it, naming
            // everything that was asked for.
            if media_handler.is_none()
                && declared_media_type(db_plan, api_request).is_none()
                && !api_request
                    .accept_media_types
                    .iter()
                    .any(renders_without_a_handler)
            {
                return Err(postrust_core::Error::InvalidMediaType(
                    api_request
                        .accept_media_types
                        .iter()
                        .map(postrust_core::api_request::MediaType::to_mime)
                        .collect::<Vec<_>>()
                        .join(", "),
                ));
            }

            // Carry the parent's whole row out of the inner query when a
            // computed relationship needs it as an argument. It has to be
            // taken here, where the real table is still in scope: a bare
            // reference to the table yields its composite type, whereas the
            // derived table it becomes one level up yields a `record`.
            let needs_parent_row = added_join_columns.iter().any(|c| c == PARENT_ROW_COLUMN)
                || matches!(&media_handler, Some((_, _, true, _)));
            let db_plan = &if needs_parent_row {
                let mut adjusted = db_plan.clone();
                // A function's result is a relation too, so the row is
                // reachable there by exactly the same means -- which is what
                // makes a computed relationship work on
                // `/rpc/getallvideogames` and not only on a table.
                //
                // Named after the table it returns, not after the function.
                // The builder aliases the call to that table (`FROM
                // test.getallvideogames() AS videogames`), because a computed
                // *column* has to be able to name the row it is a function of.
                // Naming the function here instead asks for a relation that is
                // no longer in scope, and the request that used to work --
                // `?select=name,designer:computed_designers(name)` -- answers
                // `column "getallvideogames" does not exist`.
                let (tree, relation) = match &mut adjusted {
                    DbActionPlan::Read(tree) => {
                        let relation = tree.root.from.name.clone();
                        (Some(tree), relation)
                    }
                    DbActionPlan::Call { read, .. } => {
                        let relation = read
                            .as_ref()
                            .map(|tree| tree.root.from.name.clone())
                            .unwrap_or_default();
                        (read.as_mut(), relation)
                    }
                    // The row is taken inside the statement, by `RETURNING`,
                    // and read back off the CTE as an ordinary column -- so
                    // unlike the two above there is no relation to name here,
                    // and the read tree selects the column rather than a row.
                    DbActionPlan::MutateRead { mutate, read } => {
                        let returning = match mutate {
                            postrust_core::plan::MutatePlan::Insert { returning, .. }
                            | postrust_core::plan::MutatePlan::Update { returning, .. }
                            | postrust_core::plan::MutatePlan::Delete { returning, .. } => {
                                returning
                            }
                        };
                        if !returning.iter().any(|c| c == PARENT_ROW_COLUMN) {
                            returning.push(PARENT_ROW_COLUMN.to_string());
                        }
                        // A mutation that returns no body has no read tree,
                        // and nothing to embed into either.
                        if let Some(read) = read.as_mut() {
                            if !read.root.select.iter().any(|field| {
                                field.alias.as_deref() == Some(PARENT_ROW_COLUMN)
                                    || field.field.name == PARENT_ROW_COLUMN
                            }) {
                                read.root.select.push(
                                    postrust_core::plan::CoercibleSelectField::simple(
                                        PARENT_ROW_COLUMN,
                                        "",
                                    ),
                                );
                            }
                        }
                        (None, String::new())
                    }
                };

                if let Some(tree) = tree {
                    // An empty select means every column; naming the row would
                    // otherwise replace them rather than join them.
                    if tree.root.select.is_empty() {
                        tree.root
                            .select
                            .push(postrust_core::plan::CoercibleSelectField::simple("*", ""));
                    }

                    let mut field =
                        postrust_core::plan::CoercibleField::simple(relation, String::new());
                    field.full_row = true;
                    tree.root
                        .select
                        .push(postrust_core::plan::CoercibleSelectField {
                            field,
                            aggregate: None,
                            aggregate_cast: None,
                            cast: None,
                            alias: Some(PARENT_ROW_COLUMN.to_string()),
                        });
                }
                adjusted
            } else {
                db_plan.clone()
            };

            // The `Location` of a created row is built from its primary key,
            // which the client may not have selected. The columns are added
            // under names of our own so they cannot collide with anything the
            // request asked for, and are taken back out of the response once
            // the header is built.
            //
            // Only for `Prefer: return=headers-only`, which is the request
            // that has no other way to learn where the row went. A client
            // given the row back can read the key out of it, and PostgREST
            // sends no `Location` there -- so neither does this. Deciding it
            // here also means the key columns are never added to a select
            // that had no use for them.
            let location_keys: Vec<String> = match db_plan {
                DbActionPlan::MutateRead {
                    mutate: postrust_core::plan::MutatePlan::Insert { pk_cols, .. },
                    ..
                } if !is_upsert(api_request) && wants_location(api_request) => pk_cols.clone(),
                _ => Vec::new(),
            };
            // Whether the write created a row or merged into an existing one.
            // The statement is the only place that can say, so the answer is
            // carried out of it as a column and taken back out of the response.
            let reports_inserted = matches!(
                db_plan,
                DbActionPlan::MutateRead {
                    mutate: postrust_core::plan::MutatePlan::Insert {
                        on_conflict: Some(_),
                        reports_inserted: true,
                        ..
                    },
                    ..
                }
            );

            let db_plan = &if location_keys.is_empty() && !reports_inserted {
                db_plan.clone()
            } else {
                let mut adjusted = db_plan.clone();
                if let DbActionPlan::MutateRead {
                    read: Some(tree), ..
                } = &mut adjusted
                {
                    for key in &location_keys {
                        tree.root.select.push(
                            postrust_core::plan::CoercibleSelectField::with_alias(
                                key,
                                "",
                                &format!("{}{}", LOCATION_KEY_PREFIX, key),
                            ),
                        );
                    }
                    if reports_inserted {
                        tree.root
                            .select
                            .push(postrust_core::plan::CoercibleSelectField::simple(
                                postrust_core::query::INSERTED_COLUMN,
                                "boolean",
                            ));
                    }
                }
                adjusted
            };

            // Build SQL
            let query = postrust_core::query::build_query(&ActionPlan::Db(db_plan.clone()))?;

            if !query.has_main() {
                return Ok((QueryResult::default(), None));
            }

            let (mut sql, params) = query.build_main();

            // A `Prefer: count` needs the same query without its page: the
            // total is what the filters match, not what this page returned.
            // LIMIT and OFFSET render as literals, so dropping them leaves the
            // placeholders and their numbering untouched and the count query
            // can be bound with the very same parameters.
            let mut count_sql = match api_request.preferences.count {
                Some(_) => unpaged_sql(db_plan)?,
                None => None,
            };

            // Embed related resources in the same query.
            //
            // Each relation becomes a correlated subselect in the SELECT list,
            // so PostgreSQL builds the nested JSON while it already has the
            // parent row -- one round trip for the whole tree instead of one per
            // relation per level, and no grouping or attaching in this process.
            //
            // Parent columns stay ordinary typed columns, so they are converted
            // to JSON by the same code as a request without embeds and a NUMERIC
            // or timestamp renders identically either way.
            let mut embed_filters = EmbedFilters {
                filters: &api_request.query_params.filters,
                orders: &api_request.query_params.order,
                ranges: &api_request.query_params.ranges,
                logic: &api_request.query_params.logic,
                params: Vec::new(),
                base: params.len(),
                max_rows: api_request.max_rows,
                alias_counter: 0,
                // A schema-declared handler produces the whole body from
                // these rows, so each column reaches the client as *text* --
                // and that is the one place the cast is observable. Keyed on
                // whether there is a handler at all, not on which row shape it
                // wants: `{"id":1}` against PostgREST's `{"id": 1}` is the
                // difference either way.
                rendering: match &media_handler {
                    Some(_) => postrust_core::EmbedRendering::Normalised,
                    None => postrust_core::EmbedRendering::Verbatim,
                },
            };

            // The single-query form cannot express a spread, whose columns have
            // to land in the parent object rather than under a key of their
            // own. One anywhere in the tree sends the whole request down the
            // two-query path, which can: taking the single-query path for the
            // plain relations would embed those and drop the spread, since the
            // two-query path only runs when the first produced nothing.
            let embed_parent = {
                let schema_cache = state.schema_cache().await;
                read_target(api_request, Some(db_plan), &schema_cache)
            };
            let embed_level = match embed_parent.clone() {
                Some(parent_qi)
                    if api_request.query_params.select.iter().any(|item| {
                        matches!(
                            item,
                            postrust_core::api_request::SelectItem::Relation { .. }
                                | postrust_core::api_request::SelectItem::SpreadRelation { .. }
                        )
                    }) =>
                {
                    let schema_cache = state.schema_cache().await;
                    // A computed relationship takes the parent's row, which
                    // now reaches here on every path: a read names the table,
                    // and a mutation returns the row out of the statement
                    // where it still has the table's type rather than the
                    // CTE's anonymous one.
                    let parent_row = PARENT_ROW_COLUMN_REF;
                    build_embed_expressions(
                        &schema_cache,
                        &parent_qi,
                        "src",
                        parent_row,
                        &api_request.query_params.select,
                        &mut embed_filters,
                        &[],
                        &[],
                    )?
                }
                _ => EmbedLevel::default(),
            };
            let mut embed_level = embed_level;

            // Ordering by an embedded column is resolved here rather than in
            // the read plan: the plan has no way to reach another table.
            if let (Some(parent_qi), DbActionPlan::Read(tree)) = (&embed_parent, db_plan) {
                let schema_cache = state.schema_cache().await;
                embed_level.orders = build_embed_orders(
                    &schema_cache,
                    parent_qi,
                    &tree.root.order,
                    &api_request.query_params.select,
                    &mut embed_filters,
                )?;
            }

            // Filter parameters are appended in the order the predicates were
            // renumbered, so placeholder N in the wrapped SQL lines up with
            // params[N - 1].
            let mut params = params;
            params.extend(embed_filters.params);

            // Built whenever there were relations at all, not only when they
            // produced expressions: `clients!inner()` contributes no column
            // but does narrow the parent, and its predicate binds parameters
            // that have already been appended -- left unapplied, the query is
            // sent with more parameters than it uses.
            if embed_level.saw_relations {
                // `!inner` removes parent rows, so it has to be applied before
                // the page is taken. Left where it is, the inner query's
                // LIMIT would run first and the join would then trim an
                // already-truncated page -- a request for one row could come
                // back empty. So the inner query is rebuilt unpaged and the
                // range moves out to the wrapper, after the join.
                //
                // Rebuilding is safe for parameter numbering: LIMIT and OFFSET
                // render as literals, so dropping them leaves the placeholders
                // and their count untouched.
                //
                // Ordering by an embedded resource's column needs the same
                // treatment for the same reason: the value being ordered on is
                // only reachable out here, so the page cannot be taken before
                // it is known.
                let orders_by_embed = matches!(db_plan, DbActionPlan::Read(tree)
                    if tree.root.order.iter().any(|t| t.relation.is_some()));

                let mut page = None;
                if !embed_level.inner_joins.is_empty() || orders_by_embed {
                    if let DbActionPlan::Read(tree) = db_plan {
                        let mut unpaged = tree.clone();
                        page = Some(std::mem::take(&mut unpaged.root.range));
                        sql = postrust_core::query::build_query(&ActionPlan::Db(
                            DbActionPlan::Read(unpaged),
                        ))?
                        .build_main()
                        .0;
                    }
                }

                let mut projection = String::from("src.*");
                for (field_name, expression) in &embed_level.expressions {
                    projection.push_str(", ");
                    projection.push_str(expression);
                    projection.push_str(" AS ");
                    projection.push_str(&postrust_sql::escape_ident(field_name));
                }
                // A data-modifying `WITH` has to stay at the top level:
                // PostgreSQL refuses one nested inside a subquery, which is
                // where wrapping the whole statement would put it. The clause
                // is lifted out and put back in front of the wrapper.
                let inner_sql = std::mem::take(&mut sql);
                let (with_clause, inner_sql) = split_leading_cte(&inner_sql);
                sql = format!(
                    "{}SELECT {} FROM ({}) AS src",
                    with_clause, projection, inner_sql
                );

                // `?clients=is.null` asks whether the embed matched anything,
                // not about a column. The embed's expression is the thing to
                // test, and it is only in scope out here -- referring to the
                // name it is given would be referring to a column of the very
                // select that defines it.
                let mut predicates = embed_level.inner_joins.clone();
                for filter in &api_request.query_params.filters_root {
                    let Some((_, exists)) = embed_level
                        .filterable
                        .iter()
                        .find(|(name, _)| name == &filter.field.name)
                    else {
                        continue;
                    };
                    let negated = match &filter.op_expr.operation {
                        postrust_core::api_request::Operation::Is(
                            postrust_core::api_request::IsValue::Null,
                        ) => filter.op_expr.negated,
                        postrust_core::api_request::Operation::Is(
                            postrust_core::api_request::IsValue::NotNull,
                        ) => !filter.op_expr.negated,
                        _ => continue,
                    };
                    // `is.null` asks for parents the embed did not match, so
                    // the existence test is the negation of it.
                    predicates.push(match negated {
                        true => exists.clone(),
                        false => format!("NOT {}", exists),
                    });
                }

                // `or=(clientinfo.not.is.null,contact.not.is.null)` asks the
                // same question of several embeds at once, and the answer is
                // one predicate combining their existence tests. The read
                // plan left the tree alone precisely so that it could be
                // built here, where the embeds are in scope.
                let embeds =
                    postrust_core::api_request::embedded_names(&api_request.query_params.select);
                for (path, tree) in &api_request.query_params.logic {
                    if !path.is_empty() || !tree.names_only(&embeds) {
                        continue;
                    }
                    if let Some(predicate) = embed_logic_predicate(tree, &embed_level.filterable) {
                        predicates.push(predicate);
                    }
                }

                if !predicates.is_empty() {
                    sql.push_str(" WHERE ");
                    sql.push_str(&predicates.join(" AND "));
                }

                if !embed_level.orders.is_empty() {
                    sql.push_str(" ORDER BY ");
                    sql.push_str(&embed_level.orders.join(", "));
                }

                // `!inner` and `?rel=is.null` decide which parent rows survive,
                // so the count has to be taken after them -- the same wrapper,
                // over the unpaged query.
                count_sql = count_sql.map(|base| {
                    let (with_clause, base) = split_leading_cte(&base);
                    let mut counted = format!(
                        "{}SELECT {} FROM ({}) AS src",
                        with_clause, projection, base
                    );
                    if !predicates.is_empty() {
                        counted.push_str(" WHERE ");
                        counted.push_str(&predicates.join(" AND "));
                    }
                    counted
                });

                if let Some(range) = page {
                    if let Some(limit) = range.limit {
                        sql.push_str(&format!(" LIMIT {}", limit));
                    }
                    if range.offset > 0 {
                        sql.push_str(&format!(" OFFSET {}", range.offset));
                    }
                }
            }

            // `application/geo+json` has a rendering of its own even where no
            // schema declares one: a FeatureCollection whose geometry is the
            // table's geometry column and whose properties are the rest. A
            // schema-declared handler overrides it, which is why this only
            // runs when none matched.
            //
            // Asking for GeoJSON from rows that carry no geometry is an error
            // rather than a plain JSON body: the client asked for features,
            // and a list of objects with no geometry is not features.
            let wants_geojson = media_handler.is_none()
                && api_request
                    .accept_media_types
                    .iter()
                    .any(|m| matches!(m, postrust_core::api_request::MediaType::GeoJson));
            let geojson_column = if wants_geojson {
                let schema_cache = state.schema_cache().await;
                let target = read_target(api_request, Some(db_plan), &schema_cache);
                match target.as_ref().and_then(|qi| schema_cache.get_table(qi)) {
                    Some(table) => match geometry_key(table, &api_request.query_params.select) {
                        Some(column) => Some(column),
                        None => return Err(postrust_core::Error::MissingGeometryColumn),
                    },
                    // Nothing this side can name a geometry column on -- a
                    // scalar function, say. Left to the database, which is
                    // where the answer was coming from anyway.
                    None => None,
                }
            } else {
                None
            };

            if let Some((media_type, aggregate, over_row, _)) = &media_handler {
                // The aggregate takes the table's own row type. A derived
                // table yields `record`, which will not cast to it, so where
                // the handler names a table the aggregate is applied to the
                // column carrying the real row -- the same column a computed
                // relationship reads. A handler taking `anyelement` has no
                // such requirement and takes the derived row directly.
                let argument = if *over_row {
                    format!(
                        "pgrst_media.{}",
                        postrust_sql::escape_ident(PARENT_ROW_COLUMN)
                    )
                } else {
                    "pgrst_media".to_string()
                };
                // `count(*)` alongside the handler: both are aggregates over
                // the same set, and the rendered body is one value that has
                // forgotten how many rows went into it. `Content-Range`
                // describes the row set rather than the rendering, so the
                // count has to be carried out of the same statement -- asking
                // again afterwards would be a second snapshot.
                let (with_clause, inner) = split_leading_cte(&sql);
                sql = format!(
                    "{}SELECT {}({})::text AS pgrst_body, count(*) AS {} \
                     FROM ({}) pgrst_media",
                    with_clause, aggregate, argument, MEDIA_ROW_COUNT_COLUMN, inner
                );
                debug!("Rendering {} via {}", media_type, aggregate);
            }

            // Onto the `db.query` span this runs in; see `db_span`.
            if state.config().otel_db_statement {
                tracing::Span::current().record("db.query.text", sql.as_str());
            }
            debug!("Executing SQL: {}", sql);
            debug!("With {} parameters", params.len());

            // Everything for this request runs in one transaction.
            //
            // `SET LOCAL ROLE` and `set_config(..., true)` are scoped to a
            // transaction. Sent on a bare pooled connection they apply to their
            // own implicit single-statement transaction and are gone before the
            // next statement runs, so the query executed as the pool's login
            // role with no JWT claims set -- row-level security and role grants
            // were not being applied at all. PostgreSQL says so in its log:
            // "SET LOCAL can only be used in transaction blocks".
            //
            // Embedding runs on this same transaction, which also means a
            // parent row and its children come from one snapshot rather than
            // from two separate reads.
            //
            // A transaction that cannot be begun is logged here, at error,
            // because nothing later will: the query's own logging below never
            // runs. It is the pool running dry -- every connection busy for
            // longer than `db_pool_timeout` -- or the database out of reach,
            // and either is the operator's to fix, not the client's. The
            // client is told only that the service is unavailable; the line
            // carries the cause, and the request's `request_id` and
            // `trace_id` with it.
            //
            // Timed for the pool's wait metric when a host is measuring it;
            // see `crate::observe`. A failure is timed too: a wait that ended
            // in `db_pool_timeout` is the one most worth seeing.
            let begin_timer = crate::observe::begin_timer();
            let begun = state.db().begin().await;
            crate::observe::transaction_begun(begin_timer);
            let mut tx = begun.map_err(|e| {
                error!("Could not begin a transaction: {}", e);
                postrust_core::Error::ConnectionPool(e.to_string())
            })?;

            // Make the transaction the caller's: its isolation level, read-only
            // for a read, the role and every setting a policy or function reads
            // back, and `db_pre_request`. See `Session` for each, and for the
            // order they have to go in.
            //
            // They are queued rather than run: they go out with the query, in
            // one flight where the driver can pipeline, and a failure among
            // them is the query's.
            postrust_core::Session::new(state.config(), auth.role.clone(), auth.claim_settings())
                .schema(api_request.schema.clone())
                .request(api_request.method.clone(), api_request.path.clone())
                .headers(&api_request.headers)
                .cookies(&api_request.cookies)
                .trace_id(api_request.trace_id.map(|t| t.to_string()))
                .timezone(api_request.preferences.timezone.clone())
                .read_only(is_read_only(api_request))
                .apply(&mut tx);

            // The main query. A logged failure names the driver's account of
            // it, which says more than the response is allowed to. Only faults
            // are logged at error level; see `query_error_log`.
            let log = |e: postrust_core::db::DbError| {
                match query_error_log(&e) {
                    QueryErrorLog::RefusedBefore => debug!("Refused before the query: {}", e),
                    QueryErrorLog::Refused(status) => {
                        debug!("Query refused with {}: {}", status.as_u16(), e)
                    }
                    QueryErrorLog::Fault => error!("Query error: {}", e),
                }
                postrust_core::Error::from(e)
            };

            // A rendered media type is a single value, not a row set: the
            // aggregate already produced the entire body.
            if let Some((media_type, base_type)) = media_handler
                .as_ref()
                .map(|(m, _, _, base)| (m.clone(), base.clone()))
                .or_else(|| declared_media_type(db_plan, api_request))
            {
                let raw = tx
                    .fetch_raw(&sql, &params, MEDIA_ROW_COUNT_COLUMN)
                    .await
                    .map_err(log)?;

                // The value's type is a domain this side has no mapping for,
                // so it arrives as PostgreSQL's text rendering of it. For
                // anything text-like that rendering *is* the value; for a
                // `bytea` it is `\x` and hex, which describes the bytes
                // rather than being them.
                let mut body = raw
                    .first
                    .map(|bytes| raw_column(bytes, &base_type))
                    .unwrap_or_default();

                // PostgREST reads a handler's value as raw bytes in binary
                // format, and a `jsonb` in binary format begins with a version
                // byte -- which it forwards to the client. Its own suite
                // records the result and marks it: "TODO SOH (start of
                // heading) is being added to results".
                //
                // It is a bug, and the byte makes the body invalid JSON, so it
                // is not something to do by default. Compatibility mode is
                // where this server sets out to be PostgREST rather than to be
                // right, and a client written against PostgREST's bytes gets
                // PostgREST's bytes.
                if state.config().compat_mode && base_type == "jsonb" {
                    body.insert(0, 0x01);
                }

                // A rendered body is still a window on a result set, and
                // PostgREST reports one. The row count comes from the
                // aggregate's own statement where a handler rendered the rows;
                // where the value is a function's media-type domain there was
                // no aggregate, and the rows in hand are the answer.
                let rendered_rows = match media_handler.is_some() {
                    true => raw.count.unwrap_or_default(),
                    false => raw.rows as i64,
                };
                let total = match (&api_request.preferences.count, &count_sql) {
                    (Some(preference), Some(count_sql)) => {
                        resolve_count(
                            &mut tx,
                            preference,
                            count_sql,
                            &params,
                            api_request.max_rows,
                            state.config().otel_db_statement,
                        )
                        .await?
                    }
                    _ => None,
                };
                let content_range = postrust_response::ContentRange::from_pagination(
                    top_level_offset(api_request),
                    rendered_rows,
                    total,
                );

                return Ok((
                    QueryResult {
                        status: StatusCode::OK,
                        raw_body: Some((media_type, body)),
                        content_range: Some(content_range),
                        ..QueryResult::default()
                    },
                    Some(tx),
                ));
            }

            // Embeds already came back with the parent query when the SELECT
            // list carried relations. The second-query path remains for
            // anything the single-query form did not handle -- and only
            // where there are relations to fetch, since the statements after
            // the query can go with it when there is nothing in between.
            let two_query_parent = {
                let schema_cache = state.schema_cache().await;
                read_target(api_request, Some(db_plan), &schema_cache)
            }
            .filter(|_| {
                !embed_level.saw_relations
                    && api_request.query_params.select.iter().any(|item| {
                        matches!(
                            item,
                            postrust_core::api_request::SelectItem::Relation { .. }
                                | postrust_core::api_request::SelectItem::SpreadRelation { .. }
                        )
                    })
            });

            // A function can set the response's status and headers, and only
            // it knows whether it did -- so the settings are read back after
            // it has run, on the same transaction. That is a call, a mutation
            // (through a trigger), or any request at all where a pre-request
            // function runs first: PostgREST honours what it sets on a plain
            // read too, a `Cache-Control` say.
            let reads_response_settings = is_mutation(db_plan)
                || matches!(db_plan, DbActionPlan::Call { .. })
                || state.config().db_pre_request.is_some();
            let counts = api_request.preferences.count.is_some() && count_sql.is_some();

            // The query, and with it whatever can go in the same flight. A
            // read with nothing after it commits there and then: committing
            // a read-only transaction changes nothing, so no later refusal
            // needs it open -- and one that reads its settings back reads
            // them in that flight too. A call or a write reads its settings
            // back in the same flight, unless embeds have to be fetched in
            // between.
            let ends_with_the_query =
                is_read_only(api_request) && !counts && two_query_parent.is_none();
            let (mut json_rows, response_settings, mut tx) =
                if ends_with_the_query && !reads_response_settings {
                    let rows = tx.fetch_json_and_commit(&sql, &params).await.map_err(log)?;
                    (rows, None, None)
                } else if ends_with_the_query {
                    let (rows, settings) = tx
                        .fetch_json_then_and_commit(&sql, &params, RESPONSE_SETTINGS)
                        .await
                        .map_err(log)?;
                    (rows, Some(settings), None)
                } else if reads_response_settings && two_query_parent.is_none() {
                    let (rows, settings) = tx
                        .fetch_json_then(&sql, &params, RESPONSE_SETTINGS)
                        .await
                        .map_err(log)?;
                    (rows, Some(settings), Some(tx))
                } else {
                    let rows = tx.fetch_json(&sql, &params).await.map_err(log)?;
                    (rows, None, Some(tx))
                };

            if let (Some(parent_qi), Some(tx)) = (two_query_parent, tx.as_mut()) {
                embed_relations(
                    state,
                    tx,
                    &parent_qi,
                    &api_request.query_params.select,
                    &mut json_rows,
                    api_request.max_rows,
                )
                .await?;
            }

            let (guc_headers, guc_status) =
                match (reads_response_settings, response_settings, tx.as_mut()) {
                    (false, _, _) => (None, None),
                    (true, settings, tx) => {
                        let settings = match (settings, tx) {
                            (Some(settings), _) => settings,
                            (None, Some(tx)) => tx.fetch_values(RESPONSE_SETTINGS, &[]).await?,
                            (None, None) => {
                                return Err(postrust_core::Error::Internal(
                                    "the response settings were not read".into(),
                                ))
                            }
                        };
                        let row = first_row(settings)?;
                        // Empty is absent. Once any transaction on this
                        // connection has defined a custom setting, PostgreSQL
                        // keeps the name for the rest of the session and
                        // reports it as `''` rather than null -- so a request
                        // that set nothing would otherwise inherit "the empty
                        // headers" from whichever request last set some.
                        let setting = |index: usize| {
                            row.get(index)
                                .and_then(serde_json::Value::as_str)
                                .filter(|value| !value.trim().is_empty())
                                .map(str::to_string)
                        };
                        (setting(0), setting(1))
                    }
                };

            // Both are the function's own words, so both are checked before
            // they reach the response: a malformed one is a fault in the
            // schema, and saying so is more use than a header the client
            // cannot see or a status it cannot explain.
            //
            // That includes each name and value being one HTTP can carry. The
            // response is only built after the commit, and a header it cannot
            // hold -- a newline in a value, a space in a name -- failed there,
            // where it was too late to undo the function's writes.
            if let Some(headers) = &guc_headers {
                let sendable =
                    postrust_response::parse_guc_headers(headers).is_some_and(|headers| {
                        headers.iter().all(|(name, value)| {
                            http::HeaderName::try_from(name.as_str()).is_ok()
                                && http::HeaderValue::try_from(value.as_str()).is_ok()
                        })
                    });
                if !sendable {
                    return Err(postrust_core::Error::InvalidGucHeaders);
                }
            }
            let guc_status = match guc_status {
                None => None,
                Some(status) => Some(
                    status
                        .trim()
                        .parse::<u16>()
                        .ok()
                        .and_then(|code| StatusCode::from_u16(code).ok())
                        .ok_or(postrust_core::Error::InvalidGucStatus)?,
                ),
            };

            // The count runs on the same transaction and so under the same
            // snapshot as the page it describes.
            let total = match (&api_request.preferences.count, &count_sql, tx.as_mut()) {
                (Some(preference), Some(count_sql), Some(tx)) => {
                    resolve_count(
                        tx,
                        preference,
                        count_sql,
                        &params,
                        api_request.max_rows,
                        state.config().otel_db_statement,
                    )
                    .await?
                }
                _ => None,
            };

            // Drop columns that were only selected to join the embeds.
            //
            // Except where the embed answers to that very name: embedding
            // through a column -- `?select=client_id(*)` -- keys the result by
            // the column it joined on, and the embed is what the client asked
            // for. Stripping by name would take it along with the column.
            let embed_keys: std::collections::HashSet<&str> = embed_level
                .expressions
                .iter()
                .map(|(key, _)| key.as_str())
                .collect();

            for column in added_join_columns
                .iter()
                .filter(|column| !embed_keys.contains(column.as_str()))
            {
                for row in json_rows.iter_mut() {
                    if let Some(object) = row.as_object_mut() {
                        object.remove(column);
                    }
                }
            }

            // A spread's columns belong to the object above it. This happens
            // after the join columns are dropped, not before: a spread may
            // legitimately produce a column of the same name as one added for
            // joining, and dropping it afterwards would take the wrong one.
            let spread_columns: std::collections::HashMap<String, Vec<String>> =
                embed_level.spread_columns.iter().cloned().collect();
            for row in json_rows.iter_mut() {
                flatten_spreads(row, &spread_columns);
            }

            // In PostgREST-compatibility mode, reshape RPC responses to match
            // PostgREST: un-nest the function-name-keyed column and return a
            // bare value for non-set-returning functions.
            let (mut json_rows, singular) = if state.config().compat_mode {
                if let ActionPlan::Db(DbActionPlan::Call { call, .. }) = plan {
                    unwrap_rpc_rows(json_rows, call)
                } else {
                    (json_rows, false)
                }
            } else {
                (json_rows, false)
            };

            // A function returning `void` has no result to report. PostgREST
            // answers 204 rather than a body of `null`, which is the honest
            // shape for something that returns nothing.
            let returns_void = matches!(
                db_plan,
                DbActionPlan::Call { call, .. } if call.returns_void
            );

            // `Prefer: max-affected` guards against a filter that turned out
            // to match more than the client meant. Nothing has been committed
            // yet, so refusing here undoes the whole statement.
            if let (Some(limit), postrust_core::api_request::PreferHandling::Strict) = (
                api_request.preferences.max_affected,
                &api_request.preferences.handling,
            ) {
                let affected = json_rows.len() as i64;
                let writes = is_mutation(db_plan) || matches!(db_plan, DbActionPlan::Call { .. });
                if writes && affected > limit {
                    return Err(postrust_core::Error::MaxAffectedExceeded(affected));
                }
            }

            // Any row this statement created rather than merged into. Read
            // before the column is taken back out of the response.
            let created_a_row = reports_inserted
                && json_rows.iter().any(|row| {
                    row.get(postrust_core::query::INSERTED_COLUMN)
                        == Some(&serde_json::Value::Bool(true))
                });
            if reports_inserted {
                for row in json_rows.iter_mut() {
                    if let Some(object) = row.as_object_mut() {
                        object.remove(postrust_core::query::INSERTED_COLUMN);
                    }
                }
            }

            // A `PUT` writes the row its URL names, and exactly that row. The
            // filters were applied to the body, so a body naming some other
            // row wrote nothing -- and one naming several wrote several. Both
            // are the same mistake, and PostgreSQL cannot report it because
            // neither is a database error.
            if is_upsert(api_request) && json_rows.len() != 1 {
                return Err(postrust_core::Error::PutMatchingPk);
            }

            // The created row's own address, taken from its key and then
            // removed from the body -- the client asked for a `?select=`, not
            // for the key.
            let location = build_location(api_request, &mut json_rows, &location_keys);

            // A mutation returns the affected rows only when the caller asked
            // for them; otherwise the body is empty whatever the status.
            let omit_body = is_mutation(db_plan) && !wants_representation(api_request);
            // Taken before the rows are dropped: how many a write touched is
            // still the answer to `Content-Range` and to a request for a
            // single object.
            let affected = json_rows.len();
            let rows = if omit_body { Vec::new() } else { json_rows };

            // `application/geo+json` is a rendering of these same rows, done
            // here rather than in SQL so that everything a mutation reports
            // has already happened to them: the status derived, the location
            // taken from the key, the bookkeeping columns removed. Built in
            // SQL it saw the row as the statement left it, and leaked
            // `pgrst_inserted` into a feature's properties.
            let geojson_body = geojson_column
                .as_ref()
                .filter(|_| !omit_body)
                .map(|column| {
                    (
                        "application/geo+json".to_string(),
                        feature_collection(&rows, column),
                    )
                });

            // PostgREST reports the returned window on every successful data
            // response -- reads, mutations and RPC alike, but not on errors or
            // OPTIONS. Without an exact count the total is unknown, which
            // renders as `*` and keeps the status at 200.
            // The window starts where the request asked it to, and `offset`
            // overrides the `Range` header exactly as it does when the plan
            // resolves the range -- otherwise the reported window says 0 for a
            // request that plainly skipped rows.
            let offset = top_level_offset(api_request);

            // What a mutation reports is not a window on a result set. An
            // insert or a delete reports no window at all -- `*` -- because
            // the rows it wrote are not a page of anything; an update reports
            // how many it changed; and an upsert reports nothing, the client
            // having named the row itself. PostgREST draws exactly these
            // lines, and a client reading the header expects them.
            use postrust_core::api_request::Mutation;
            let content_range = match mutation_kind(api_request) {
                Some(Mutation::SingleUpsert) => None,
                Some(Mutation::Create) | Some(Mutation::Delete) => {
                    Some(postrust_response::ContentRange::new(1, 0, total))
                }
                Some(Mutation::Update) => Some(postrust_response::ContentRange::new(
                    0,
                    affected as i64 - 1,
                    total,
                )),
                None => Some(postrust_response::ContentRange::from_pagination(
                    offset,
                    rows.len() as i64,
                    total,
                )),
            };

            // An offset past the end of the result is a range the server
            // cannot satisfy, and saying how many rows there actually are is
            // what lets the client correct it. Only reachable with a count
            // preference: without one the total is unknown and there is
            // nothing to be past the end of.
            if content_range
                .as_ref()
                .is_some_and(|range| range.status() == StatusCode::RANGE_NOT_SATISFIABLE)
                && mutation_kind(api_request).is_none()
            {
                return Err(postrust_core::Error::InvalidRange(format!(
                    "An offset of {} was requested, but there are only {} rows.",
                    offset,
                    total.unwrap_or(0)
                )));
            }

            let result = QueryResult {
                // A status the function set outright wins: it knows things
                // about the outcome that the request's shape cannot say.
                status: match (guc_status, returns_void) {
                    (Some(status), _) => status,
                    (None, true) => StatusCode::NO_CONTENT,
                    (None, false) => {
                        mutation_status(db_plan, api_request, created_a_row, reports_inserted)
                            .unwrap_or_else(|| {
                                content_range
                                    .as_ref()
                                    .map(postrust_response::ContentRange::status)
                                    .unwrap_or(StatusCode::OK)
                            })
                    }
                },
                rows,
                singular,
                omit_body: omit_body || returns_void,
                affected,
                location,
                content_range,
                guc_headers,
                raw_body: geojson_body,
                ..Default::default()
            };
            // Committed by the caller, once the response is known to be one
            // it can send: refusals still to come undo the write.
            Ok((result, tx))
        }
        ActionPlan::Info(info_plan) => {
            use postrust_core::plan::InfoPlan;

            // Return appropriate metadata based on the info type
            let response_data = match info_plan {
                InfoPlan::OpenApiSpec => {
                    // Return basic server info for root endpoint
                    serde_json::json!({
                        "name": "postrust",
                        "version": env!("CARGO_PKG_VERSION"),
                        "description": "PostgREST-compatible REST API for PostgreSQL"
                    })
                }
                InfoPlan::RelationInfo(qi) => {
                    serde_json::json!({
                        "schema": qi.schema,
                        "name": qi.name,
                        "type": "relation"
                    })
                }
                InfoPlan::RoutineInfo(qi) => {
                    serde_json::json!({
                        "schema": qi.schema,
                        "name": qi.name,
                        "type": "routine"
                    })
                }
            };

            Ok((
                QueryResult {
                    status: StatusCode::OK,
                    rows: vec![response_data],
                    ..Default::default()
                },
                None,
            ))
        }
    }
}

/// How a failed query is logged.
#[derive(Debug, PartialEq, Eq)]
enum QueryErrorLog {
    /// A queued statement refused the request: a pre-request function raising
    /// to say no. Debug.
    RefusedBefore,
    /// The schema answering the request on purpose, with a 4xx: a function
    /// raising a `PT` status or `PGRST`, a constraint, a value that is not of
    /// its column's type. Debug.
    Refused(http::StatusCode),
    /// Everything else, and every connection failure. Error.
    Fault,
}

/// Which of [`QueryErrorLog`] a failed query is.
///
/// A log line at error level for every refusal buries the failures that are
/// faults: a schema that answers "not found" by raising `PT404` produced one
/// for each such request, with nothing to tell it from a broken query.
///
/// Decided by SQLSTATE, not by the status the response carries. Many faults
/// are answered with a 4xx too -- SQL the planner got wrong (`42601`,
/// `42703`), a table the schema cache still lists after it was dropped
/// (`42P01`), a missing grant (`42501`), a bare `RAISE EXCEPTION` (`P0001`) or
/// a division by zero in a function -- and each of those needs someone to
/// look at it. Only the classes that are the request meeting a rule the
/// schema states are refusals.
fn query_error_log(e: &postrust_core::db::DbError) -> QueryErrorLog {
    if e.is_connection() {
        return QueryErrorLog::Fault;
    }
    if e.was_queued() {
        return QueryErrorLog::RefusedBefore;
    }
    let Some(database) = e.as_database() else {
        return QueryErrorLog::Fault;
    };
    let status = database.status_code();
    let code = database.code.as_str();
    let refusal = match code {
        // Raised on purpose to answer with a status.
        "PGRST" => true,
        _ if code.starts_with("PT") => true,
        // Integrity constraints: a duplicate key, a check, a foreign key.
        _ if code.starts_with("23") => true,
        // Data exceptions on what the client sent: a value that is not an
        // integer, a date out of range, a string too long. Not division by
        // zero, which is a function's arithmetic rather than its input.
        "22012" => false,
        _ if code.starts_with("22") => status == http::StatusCode::BAD_REQUEST,
        // A write in the read-only transaction a GET runs in, or one the
        // view cannot take: the method was the client's choice.
        "25006" => true,
        _ if database.is_unwritable_view() => true,
        _ => false,
    };
    match refusal && status.is_client_error() {
        true => QueryErrorLog::Refused(status),
        false => QueryErrorLog::Fault,
    }
}

/// A function's response settings, read back after it ran.
const RESPONSE_SETTINGS: &str = "SELECT current_setting('response.headers', true), \
                                 current_setting('response.status', true)";

/// What the resource this request names allows, or why it cannot say.
///
/// The path is resolved by the same parser the handler uses rather than by
/// splitting on `/` here, so a percent-encoded table name, an `Accept-Profile`
/// naming another schema, and the `/rpc/` prefix all mean what they mean
/// everywhere else.
pub async fn allow<S: State>(
    state: &S,
    probe: http::Request<bytes::Bytes>,
) -> Result<String, postrust_core::Error> {
    use postrust_core::api_request::Action;

    let api_request = parse_request(
        &probe,
        state.default_schema(),
        state.schemas(),
        state.config().db_max_rows,
    )?;

    let schema_cache = state.schema_cache().await;
    match &api_request.action {
        Action::RelationInfo(qi) => Ok(schema_cache.require_table(qi)?.allow()),
        Action::RoutineInfo { qi, .. } => Ok(routine_allow(schema_cache.get_routines(qi))),
        // The schema description, which is read and nothing else.
        _ => Ok("OPTIONS,GET,HEAD".to_string()),
    }
}

/// The methods a function answers, for the `Allow` of an OPTIONS.
///
/// A volatile function may change data, so it is reachable only by POST. One
/// that cannot can also be read, and is offered under GET and HEAD as well.
/// An overloaded name is readable when any of its overloads is.
fn routine_allow(routines: Option<&Vec<postrust_core::schema_cache::Routine>>) -> String {
    let readable = routines.is_some_and(|routines| {
        routines.iter().any(|routine| {
            !matches!(
                routine.volatility,
                postrust_core::schema_cache::FuncVolatility::Volatile
            )
        })
    });

    match readable {
        true => "OPTIONS,GET,HEAD,POST".to_string(),
        false => "OPTIONS,POST".to_string(),
    }
}

/// Reshape RPC rows for PostgREST-compatibility mode.
///
/// `SELECT * FROM func(...)` names its single output column after the function
/// (for scalar/`json` returns), which serializes to rows like
/// `[{"func": <value>}]`. PostgREST instead returns the bare value. This
/// un-nests that wrapper column and reports whether the result should be
/// rendered as a single (un-arrayed) value — i.e. when the function is not
/// set-returning.
///
/// The decision is driven by the plan's return-type metadata: composite and
/// `record` returns (`RETURNS TABLE`, row types) have real output columns and
/// are never un-nested, even when a single column happens to share the
/// function's name (e.g. `CREATE FUNCTION foo() RETURNS TABLE(foo int)`).
fn unwrap_rpc_rows(
    rows: Vec<serde_json::Value>,
    call: &CallPlan,
) -> (Vec<serde_json::Value>, bool) {
    let singular = !call.returns_set;

    if call.returns_composite {
        return (rows, singular);
    }

    let fname = call.function.name.as_str();
    let unwrapped = rows
        .into_iter()
        .map(|row| match row {
            serde_json::Value::Object(ref map) if map.len() == 1 && map.contains_key(fname) => {
                map.get(fname).cloned().unwrap_or(serde_json::Value::Null)
            }
            other => other,
        })
        .collect();

    (unwrapped, singular)
}

/// The table a read targets, if the request is a read.
fn read_target(
    api_request: &postrust_core::api_request::ApiRequest,
    db_plan: Option<&DbActionPlan>,
    schema_cache: &postrust_core::SchemaCache,
) -> Option<postrust_core::api_request::QualifiedIdentifier> {
    use postrust_core::api_request::{Action, DbAction};

    match &api_request.action {
        Action::Db(DbAction::RelationRead { qi, .. }) => Some(qi.clone()),
        // The rows a mutation affected are rows of that table, so `?select=`
        // reaches through them to related resources exactly as it does on a
        // read -- `DELETE /tasks?select=name,project:projects(id)` says what
        // was deleted and what it belonged to.
        Action::Db(DbAction::RelationMut { qi, .. }) => Some(qi.clone()),
        // A function returning a table's rows embeds, and renders, exactly as
        // that table does.
        //
        // Which table that is was settled when the overload was chosen: a
        // name may carry several signatures returning different things, and
        // asking the cache again by name answers for whichever comes first.
        // It also knows the answer for a return type this side cannot read
        // off the name at all, such as a domain over a table.
        Action::Db(DbAction::Routine { qi, .. }) => match db_plan {
            Some(DbActionPlan::Call { read, .. }) => {
                read.as_ref().map(|tree| tree.root.from.clone())
            }
            _ => schema_cache
                .routine_returned_table(qi)
                .map(|table| table.qualified_identifier()),
        },
        _ => None,
    }
}

/// One `?or=` or `?and=` over embedded resources, as a `WHERE` fragment.
///
/// Every leaf asks whether an embed matched, which is an existence test, and
/// the tree is what combines them. `None` where any leaf asks something else,
/// since half a condition is worse than none.
fn embed_logic_predicate(
    tree: &postrust_core::api_request::LogicTree,
    filterable: &[(String, String)],
) -> Option<String> {
    use postrust_core::api_request::{IsValue, LogicOperator, LogicTree, Operation};

    match tree {
        LogicTree::Expr {
            negated,
            op,
            children,
        } => {
            let rendered = children
                .iter()
                .map(|child| embed_logic_predicate(child, filterable))
                .collect::<Option<Vec<String>>>()?;
            let joined = rendered.join(match op {
                LogicOperator::And => " AND ",
                LogicOperator::Or => " OR ",
            });
            Some(match negated {
                true => format!("NOT ({joined})"),
                false => format!("({joined})"),
            })
        }
        LogicTree::Stmt(filter) => {
            let (_, exists) = filterable
                .iter()
                .find(|(name, _)| name == &filter.field.name)?;
            // `is.null` asks for parents the embed did not match, so the
            // existence test is the negation of it.
            let negated = match &filter.op_expr.operation {
                Operation::Is(IsValue::Null) => filter.op_expr.negated,
                Operation::Is(IsValue::NotNull) => !filter.op_expr.negated,
                _ => return None,
            };
            Some(match negated {
                true => exists.clone(),
                false => format!("NOT {exists}"),
            })
        }
    }
}

/// Round a geometry's numbers to nine decimal places.
///
/// Not cosmetic: nine is what `ST_AsGeoJSON` emits by default, and a geometry
/// read through the type's own rendering carries every digit the double has.
/// Nine decimal places of latitude is under a tenth of a millimetre, so the
/// digits past it describe floating-point representation rather than any
/// place on the earth.
fn round_coordinates(value: &mut serde_json::Value) {
    match value {
        serde_json::Value::Array(items) => items.iter_mut().for_each(round_coordinates),
        serde_json::Value::Object(fields) => fields.values_mut().for_each(round_coordinates),
        serde_json::Value::Number(number) => {
            let Some(float) = number.as_f64() else { return };
            if float.fract() == 0.0 {
                return;
            }
            // Through the decimal text rather than by arithmetic: scaling by a
            // power of ten and rounding back reintroduces exactly the digits
            // this is meant to drop.
            if let Ok(rounded) = format!("{float:.9}").parse::<f64>() {
                if let Some(rounded) = serde_json::Number::from_f64(rounded) {
                    *number = rounded;
                }
            }
        }
        _ => {}
    }
}

/// Where the response's window starts.
///
/// The same resolution the plan itself used, asked again rather than restated:
/// the JSON path and the rendered-media path both report a `Content-Range`,
/// and a `Content-Range` that disagrees with the rows underneath it is worse
/// than none at all.
fn top_level_offset(api_request: &ApiRequest) -> i64 {
    postrust_core::plan::resolve_top_level_range(api_request).offset
}

/// The first column of a row, as the bytes the client should receive.
///
/// A media type the schema declared is carried by a domain, and a domain is a
/// type this side has no decoder for -- so the value arrives as PostgreSQL's
/// own text rendering. For text, xml or json that rendering is the value. For
/// `bytea` it is `\x` followed by hex, which is a description of the bytes
/// and not the bytes, so it is decoded back.
fn raw_column(bytes: Vec<u8>, base_type: &str) -> Vec<u8> {
    if base_type != "bytea" {
        return bytes;
    }
    // Text or binary, depending on whether this connection asked for a format
    // it could name -- so the `\x` rendering is decoded where it appears and
    // the bytes are passed through where they do not.
    let Some(hex) = bytes.strip_prefix(b"\\x") else {
        return bytes;
    };
    hex.chunks(2)
        .map(|pair| {
            let digits = std::str::from_utf8(pair).ok()?;
            u8::from_str_radix(digits, 16).ok()
        })
        .collect::<Option<Vec<u8>>>()
        .unwrap_or(bytes)
}

/// Whether the selection is the one a table has when none was asked for.
///
/// A handler declared for a table renders the table's own rows, so it applies
/// only where the request did not reshape them. `?select=*` is that shape
/// written out; `?select=id` and any embed are not.
fn has_default_select(select: &[postrust_core::api_request::SelectItem]) -> bool {
    use postrust_core::api_request::SelectItem;

    match select {
        [] => true,
        [SelectItem::Field { field, alias, .. }] => field.name == "*" && alias.is_none(),
        _ => false,
    }
}

/// Whether this server renders the type itself, with nothing declared for it.
fn renders_without_a_handler(media: &postrust_core::api_request::MediaType) -> bool {
    use postrust_core::api_request::MediaType;

    matches!(
        media,
        MediaType::ApplicationJson
            | MediaType::Any
            | MediaType::TextCsv
            | MediaType::GeoJson
            | MediaType::SingularJson { .. }
            | MediaType::ArrayJson { .. }
            | MediaType::OpenApi
    )
}

/// Render rows as a GeoJSON `FeatureCollection`.
///
/// The geometry arrives already rendered: a user-defined type is rendered by
/// the database, and what PostGIS renders a geometry as is GeoJSON -- with a
/// `crs` key, which a feature's geometry does not carry. Everything else on
/// the row is the feature's properties, the geometry excepted, since a
/// feature's geometry is not also one of its properties.
fn feature_collection(rows: &[serde_json::Value], column: &str) -> Vec<u8> {
    let features: Vec<serde_json::Value> = rows
        .iter()
        .map(|row| {
            let mut properties = row.clone();
            let geometry = properties
                .as_object_mut()
                .and_then(|fields| fields.remove(column))
                .map(|mut geometry| {
                    if let Some(fields) = geometry.as_object_mut() {
                        fields.remove("crs");
                    }
                    round_coordinates(&mut geometry);
                    geometry
                })
                .unwrap_or(serde_json::Value::Null);
            serde_json::json!({
                "type": "Feature",
                "geometry": geometry,
                "properties": properties,
            })
        })
        .collect();
    serde_json::to_vec(&serde_json::json!({
        "type": "FeatureCollection",
        "features": features,
    }))
    .unwrap_or_default()
}

/// The key under which a GeoJSON feature's geometry is read.
///
/// A table may carry more than one geometry column, and `?select=` is what
/// says which of them the request is about: `select=id,name,range_area` asks
/// for a collection of areas, not for whichever geometry the catalogue
/// happens to list first. The name wanted is the one the response uses, so an
/// aliased column is found under its alias -- which is also the key that has
/// to come back out of `properties`, since a feature's geometry is not also
/// one of its properties.
///
/// `None` means the rows carry no geometry at all, which is not a GeoJSON
/// response with something missing from it but a request that cannot be
/// answered.
fn geometry_key(
    table: &postrust_core::schema_cache::Table,
    select: &[postrust_core::api_request::SelectItem],
) -> Option<String> {
    use postrust_core::api_request::SelectItem;

    let is_geometry = |name: &str| {
        table
            .get_column(name)
            .is_some_and(|c| matches!(c.nominal_type.as_str(), "geometry" | "geography"))
    };

    // `*` and a named column can both appear in one selection, and a column
    // named outright is the one meant even so: `select=*,range_area` is a
    // request about areas.
    let mut selects_every_column = select.is_empty();
    for item in select {
        if let SelectItem::Field { field, alias, .. } = item {
            if field.name == "*" {
                selects_every_column = true;
            } else if is_geometry(&field.name) {
                return Some(alias.clone().unwrap_or_else(|| field.name.clone()));
            }
        }
    }

    if !selects_every_column {
        return None;
    }
    table
        .columns
        .values()
        .find(|c| matches!(c.nominal_type.as_str(), "geometry" | "geography"))
        .map(|c| c.name.clone())
}

/// Whether any part of the selection, at any depth, asks for an aggregate.
fn selects_an_aggregate(api_request: &ApiRequest) -> bool {
    use postrust_core::api_request::SelectItem;

    fn walk(items: &[SelectItem]) -> bool {
        items.iter().any(|item| match item {
            SelectItem::Field { aggregate, .. } => aggregate.is_some(),
            SelectItem::Relation { select, .. } => walk(select),
            SelectItem::SpreadRelation { select, .. } => walk(select),
        })
    }

    walk(&api_request.query_params.select)
}

/// Whether this request is a read, and so may run read-only.
///
/// It is the method that decides, not the plan: calling a function is a read
/// over GET and a write over POST, and the same function may be either. Asking
/// the plan instead would class every call as a read and refuse the writes a
/// POST is entitled to make.
fn is_read_only(api_request: &ApiRequest) -> bool {
    use postrust_core::api_request::{Action, DbAction, InvokeMethod};

    match &api_request.action {
        Action::Db(DbAction::RelationRead { .. }) | Action::Db(DbAction::SchemaRead { .. }) => true,
        Action::Db(DbAction::Routine { invoke_method, .. }) => {
            matches!(invoke_method, InvokeMethod::InvRead { .. })
        }
        Action::Db(DbAction::RelationMut { .. }) => false,
        // Metadata only; nothing is executed against the database.
        Action::RelationInfo(_) | Action::RoutineInfo { .. } | Action::SchemaInfo => true,
    }
}

/// The plan's SQL with its page removed, for counting what the filters match.
///
/// Only reads and calls are counted. A mutation's `Content-Range` reports the
/// rows it affected, which is the result set itself.
fn unpaged_sql(
    db_plan: &postrust_core::plan::DbActionPlan,
) -> Result<Option<String>, postrust_core::Error> {
    use postrust_core::plan::{ActionPlan, DbActionPlan};

    let unpaged = match db_plan {
        DbActionPlan::Read(tree) => {
            let mut tree = tree.clone();
            tree.root.range = Default::default();
            DbActionPlan::Read(tree)
        }
        DbActionPlan::Call { call, read } => {
            let read = read.as_ref().map(|tree| {
                let mut tree = tree.clone();
                tree.root.range = Default::default();
                tree
            });
            DbActionPlan::Call {
                call: call.clone(),
                read,
            }
        }
        DbActionPlan::MutateRead { .. } => return Ok(None),
    };

    let query = postrust_core::query::build_query(&ActionPlan::Db(unpaged))?;
    Ok(query.has_main().then(|| query.build_main().0))
}

/// The total the `Prefer: count` asked for, or `None` when it asked for none.
///
/// `exact` counts the rows the filters match. `planned` asks the query planner
/// what it expects without running anything, which is cheap on a large table
/// and correspondingly rough. `estimated` is the planner's guess only where it
/// matters: below the server's row ceiling the exact count is already paid for,
/// so it is used, and above it the larger of the two is reported -- an estimate
/// that came in under a count we have actually taken would be plainly wrong.
async fn resolve_count<T: Tx>(
    tx: &mut T,
    preference: &postrust_core::api_request::PreferCount,
    count_sql: &str,
    params: &[postrust_sql::SqlParam],
    max_rows: Option<i64>,
    record_statement: bool,
) -> Result<Option<i64>, postrust_core::Error> {
    use postrust_core::api_request::PreferCount;

    let text = record_statement;
    match preference {
        PreferCount::Exact => exact_count(tx, count_sql, params, text).await,
        PreferCount::Planned => planned_count(tx, count_sql, params, text).await,
        PreferCount::Estimated => match (exact_count(tx, count_sql, params, text).await?, max_rows)
        {
            (Some(exact), Some(ceiling)) if exact <= ceiling => Ok(Some(exact)),
            (Some(exact), _) => {
                let planned = planned_count(tx, count_sql, params, text).await?;
                Ok(Some(exact.max(planned.unwrap_or(exact))))
            }
            (None, _) => planned_count(tx, count_sql, params, text).await,
        },
    }
}

/// The number of rows the filters match.
///
/// `record_statement` is `otel_db_statement`: whether the statement goes on
/// its span as `db.query.text`, as the main one does on `db.query`.
async fn exact_count<T: Tx>(
    tx: &mut T,
    count_sql: &str,
    params: &[postrust_sql::SqlParam],
    record_statement: bool,
) -> Result<Option<i64>, postrust_core::Error> {
    let sql = format!("SELECT count(*) FROM ({}) AS pgrst_count", count_sql);
    let span = count_span("count");
    if record_statement {
        span.record("db.query.text", sql.as_str());
    }
    let row = first_row(tx.fetch_values(&sql, params).instrument(span).await?)?;
    Ok(row.first().and_then(serde_json::Value::as_i64))
}

/// The number of rows the query planner expects, without running the query.
async fn planned_count<T: Tx>(
    tx: &mut T,
    count_sql: &str,
    params: &[postrust_sql::SqlParam],
    record_statement: bool,
) -> Result<Option<i64>, postrust_core::Error> {
    let sql = format!("EXPLAIN (FORMAT JSON) {}", count_sql);
    let span = count_span("explain");
    if record_statement {
        span.record("db.query.text", sql.as_str());
    }
    let plan = first_row(tx.fetch_values(&sql, params).instrument(span).await?)?
        .into_iter()
        .next()
        .unwrap_or(serde_json::Value::Null);
    Ok(plan
        .get(0)
        .and_then(|node| node.get("Plan"))
        .and_then(|node| node.get("Plan Rows"))
        .and_then(serde_json::Value::as_i64))
}

/// Whether this plan changes data.
fn is_mutation(db_plan: &postrust_core::plan::DbActionPlan) -> bool {
    matches!(
        db_plan,
        postrust_core::plan::DbActionPlan::MutateRead { .. }
    )
}

/// Whether the caller asked for the affected rows back.
fn wants_representation(api_request: &ApiRequest) -> bool {
    matches!(
        api_request.preferences.representation,
        postrust_core::api_request::PreferRepresentation::Full
    )
}

/// Whether a created row's address is worth reporting.
///
/// `Prefer: return=headers-only` asks for exactly that and nothing else, and
/// is the only request PostgREST answers with a `Location`. A caller taking
/// the row back reads the key out of the body; a caller wanting neither asked
/// for a minimal response and gets one.
fn wants_location(api_request: &ApiRequest) -> bool {
    matches!(
        api_request.preferences.representation,
        postrust_core::api_request::PreferRepresentation::HeadersOnly
    )
}

/// The status PostgREST gives a successful mutation, or `None` for anything
/// that isn't one.
///
/// An insert reports 201 Created. An update or delete reports 204 No Content
/// unless the caller asked for the rows back, in which case there is content
/// to report and it is a plain 200.
fn mutation_status(
    db_plan: &postrust_core::plan::DbActionPlan,
    api_request: &ApiRequest,
    created_a_row: bool,
    // Whether the statement was in a position to say. A write with no conflict
    // clause could only have created rows, and one on a relation that will not
    // give up `xmax` cannot tell either way -- in both cases there is nothing
    // to report but the creation.
    could_have_merged: bool,
) -> Option<StatusCode> {
    use postrust_core::api_request::PreferResolution;
    use postrust_core::plan::{DbActionPlan, MutatePlan};

    let DbActionPlan::MutateRead { mutate, .. } = db_plan else {
        return None;
    };

    let merging = matches!(
        api_request.preferences.resolution,
        Some(PreferResolution::MergeDuplicates)
    );

    Some(match mutate {
        // A `PUT` names the row it is writing, so from the client's side
        // nothing was discovered by the response -- it reports the outcome
        // only where it asked to see the row.
        MutatePlan::Insert { .. } if is_upsert(api_request) => {
            match (
                wants_representation(api_request),
                created_a_row || !could_have_merged,
            ) {
                (false, _) => StatusCode::NO_CONTENT,
                (true, true) => StatusCode::CREATED,
                (true, false) => StatusCode::OK,
            }
        }
        // A `POST` that merged into every row it touched created nothing, and
        // 201 would be a promise that something new is at the `Location`.
        MutatePlan::Insert { .. } if merging && could_have_merged && !created_a_row => {
            StatusCode::OK
        }
        MutatePlan::Insert { .. } => StatusCode::CREATED,
        _ => match wants_representation(api_request) {
            true => StatusCode::OK,
            false => StatusCode::NO_CONTENT,
        },
    })
}

/// Ensure every embedded relation's join column is selected on the parent.
///
/// Returns the columns that were added purely to make the join possible, so
/// they can be removed from the response afterwards. An empty select list means
/// "all columns", in which case nothing needs adding.
fn add_embed_join_columns(
    api_request: &mut postrust_core::api_request::ApiRequest,
    schema_cache: &postrust_core::SchemaCache,
) -> Result<Vec<String>, postrust_core::Error> {
    use postrust_core::api_request::{Field, SelectItem};

    let Some(parent_qi) = read_target(api_request, None, schema_cache) else {
        return Ok(Vec::new());
    };

    let select = &api_request.query_params.select;
    if select.is_empty() {
        return Ok(Vec::new());
    }

    // `*` already names every column, so no join column needs adding -- and
    // adding one anyway selects it twice, which makes `src.*` ambiguous. The
    // relations are still walked: a computed relationship needs the parent's
    // row carried out, and `*` does not provide that.
    let selects_everything = select
        .iter()
        .any(|item| matches!(item, SelectItem::Field { field, .. } if field.name == "*"));

    // Only a column selected under its own name serves as a join key. An
    // aliased one -- `myId:id` -- arrives in the result as `myId`, so the
    // correlation would look for a column the inner query no longer has; the
    // same goes for one that was cast or reached into with a JSON path.
    let selected: std::collections::HashSet<String> = select
        .iter()
        .filter_map(|item| match item {
            SelectItem::Field {
                field,
                aggregate: None,
                cast: None,
                alias: None,
                ..
            } if field.json_path.is_empty() => Some(field.name.clone()),
            _ => None,
        })
        .collect();

    let mut added = Vec::new();

    for item in select.clone() {
        // A spread joins on a parent column exactly as a plain embed does, so
        // it needs that column selected just the same. Without it the parent
        // row has no key to match children against and every spread column
        // comes back null.
        let (relation, hint) = match &item {
            SelectItem::Relation { relation, hint, .. } => (relation, hint),
            SelectItem::SpreadRelation { relation, hint, .. } => (relation, hint),
            SelectItem::Field { .. } => continue,
        };

        let rel = schema_cache
            .find_relationship(&parent_qi, relation, hint.as_deref(), &parent_qi.schema)?
            .ok_or_else(|| {
                schema_cache.relationship_not_found(
                    &parent_qi,
                    relation,
                    hint.as_deref(),
                    &parent_qi.schema,
                )
            })?;

        let plan = postrust_core::embed::EmbedPlan::resolve(rel, schema_cache)?;

        // A computed relationship joins on nothing -- the parent row is the
        // function's argument. That row has to be carried out of the inner
        // query, which is a column of its own rather than a join key. After a
        // mutation it is carried out of the statement itself, by `RETURNING`,
        // because that is the last place the row still has the table's type.
        if plan.function.is_some() {
            if !added.iter().any(|c| c == PARENT_ROW_COLUMN) {
                added.push(PARENT_ROW_COLUMN.to_string());
            }
            continue;
        }

        // Every column the join is on, not just the first: a foreign key over
        // several columns correlates on all of them, and one left unselected
        // is a column the correlation cannot see.
        let join_columns: Vec<String> = match plan.columns.is_empty() {
            true => vec![plan.local_column.clone()],
            false => plan
                .columns
                .iter()
                .map(|(local, _)| local.clone())
                .collect(),
        };

        if selects_everything {
            continue;
        }

        for column in join_columns {
            if !selected.contains(&column) && !added.contains(&column) {
                api_request.query_params.select.push(SelectItem::Field {
                    field: Field::simple(&column),
                    aggregate: None,
                    aggregate_cast: None,
                    cast: None,
                    alias: None,
                });
                added.push(column);
            }
        }
    }

    Ok(added)
}

/// Build the SELECT-list expressions that embed relations in one query.
///
/// Returns one `(field_name, expression)` per requested relation. Each
/// expression is a correlated subselect yielding JSON, so the whole tree comes
/// back from the parent query rather than from a query per relation per level.
///
/// `alias_counter` hands out a distinct alias per level, so a self-referential
/// relationship stays unambiguous.
/// Filters addressed at embedded resources, and the parameters they bind.
///
/// The embed expressions are assembled as SQL text and wrapped around a main
/// query that already owns `$1..$n`, so any placeholder introduced here has to
/// be renumbered past that point and its value appended in the same order.
struct EmbedFilters<'a> {
    /// Every path-scoped filter on the request, e.g. `(["clients"], id=eq.1)`.
    filters: &'a [(
        postrust_core::api_request::EmbedPath,
        postrust_core::api_request::Filter,
    )],
    /// Values bound by the predicates built so far, in placeholder order.
    params: Vec<postrust_sql::SqlParam>,
    /// How many parameters the main query already uses.
    base: usize,
    /// Ordering asked of each embedded resource, by path.
    orders: &'a [(
        postrust_core::api_request::EmbedPath,
        Vec<postrust_core::api_request::OrderTerm>,
    )],
    /// Ranges asked of each embedded resource, by dotted path.
    ranges: &'a std::collections::HashMap<String, postrust_core::api_request::Range>,
    /// `and=`/`or=` groups asked of each embedded resource, by path.
    logic: &'a [(
        postrust_core::api_request::EmbedPath,
        postrust_core::api_request::LogicTree,
    )],
    /// Row cap applied to each embedded resource.
    max_rows: Option<i64>,
    /// Source of unique subquery aliases across the whole embed tree.
    alias_counter: usize,
    /// How an embedded resource is rendered.
    ///
    /// `Verbatim` everywhere except under a media handler that stringifies the
    /// whole row: there each column is rendered as *text*, and a `json` column
    /// renders exactly as written where PostgREST's `jsonb` one renders
    /// normalised. See [`postrust_core::EmbedRendering`].
    rendering: postrust_core::EmbedRendering,
}

/// Renumber the placeholders in `sql` so they start after `offset`.
///
/// Scans rather than string-replaces: a naive replacement of `$1` would also
/// corrupt the `$1` inside `$10`.
fn shift_placeholders(sql: &str, offset: usize) -> String {
    let mut out = String::with_capacity(sql.len());
    let mut chars = sql.char_indices().peekable();

    while let Some((i, ch)) = chars.next() {
        if ch != '$' {
            out.push(ch);
            continue;
        }
        let start = i + 1;
        let mut end = start;
        while let Some((j, d)) = chars.peek() {
            if d.is_ascii_digit() {
                end = j + d.len_utf8();
                chars.next();
            } else {
                break;
            }
        }
        match sql[start..end].parse::<usize>() {
            Ok(n) => out.push_str(&format!("${}", n + offset)),
            Err(_) => out.push('$'),
        }
    }

    out
}

impl EmbedFilters<'_> {
    /// A fresh alias, unique across the embed tree.
    fn next_alias(&mut self) -> String {
        self.alias_counter += 1;
        format!("e{}", self.alias_counter)
    }

    /// The `WHERE` fragment for one embedded resource, or `None` if unfiltered.
    /// The row window an embedded resource was asked for.
    ///
    /// The server's own cap still applies, so an embed cannot be asked for
    /// more rows than the server is willing to return.
    fn range_for(&self, path: &[String], names: &[String]) -> (Option<i64>, i64) {
        let range = self
            .ranges
            .get(&path.join("."))
            .or_else(|| self.ranges.get(&names.join(".")));
        let requested = range.and_then(|r| r.limit);
        let offset = range.map(|r| r.offset).unwrap_or(0);
        let limit = match (requested, self.max_rows) {
            (Some(limit), Some(cap)) => Some(limit.min(cap)),
            (Some(limit), None) => Some(limit),
            (None, cap) => cap,
        };
        (limit, offset)
    }

    fn predicate_for(
        &mut self,
        path: &[String],
        names: &[String],
        child_qi: &postrust_core::api_request::QualifiedIdentifier,
        schema_cache: &postrust_core::SchemaCache,
    ) -> Result<Option<String>, postrust_core::Error> {
        let addresses = |candidate: &[String]| candidate == path || candidate == names;

        let matching: Vec<_> = self
            .filters
            .iter()
            .filter(|(filter_path, _)| addresses(filter_path))
            .map(|(_, filter)| filter.clone())
            .collect();

        let matching_logic: Vec<_> = self
            .logic
            .iter()
            .filter(|(logic_path, _)| addresses(logic_path))
            .map(|(_, tree)| tree.clone())
            .collect();

        if matching.is_empty() && matching_logic.is_empty() {
            return Ok(None);
        }

        let table = schema_cache.get_table(child_qi);
        let mut parts = Vec::with_capacity(matching.len());

        for filter in &matching {
            // Inside the child's subselect an unqualified name binds to the
            // child, but only because the child has that column: an unknown
            // name would resolve outward to the correlated parent and filter
            // the wrong table silently. Refuse instead.
            let column = table
                .and_then(|t| t.columns.get(&filter.field.name))
                .ok_or_else(|| {
                    postrust_core::Error::ColumnNotFound(format!(
                        "{}.{}",
                        child_qi.name, filter.field.name
                    ))
                })?;

            let frag =
                postrust_core::query::QueryBuilder::filter_sql(filter, &column.nominal_type)?;
            parts.push(shift_placeholders(
                frag.sql(),
                self.base + self.params.len(),
            ));
            self.params.extend(frag.params().iter().cloned());
        }

        for tree in matching_logic {
            // A name the child does not have would bind outward to the
            // correlated parent and filter the wrong table, exactly as a plain
            // filter would, so the tree is checked before it is rendered.
            let mut unknown = None;
            check_logic_columns(&tree, table, &mut unknown);
            if let Some(name) = unknown {
                return Err(postrust_core::Error::ColumnNotFound(format!(
                    "{}.{}",
                    child_qi.name, name
                )));
            }

            let resolver = |name: &str| -> String {
                table
                    .and_then(|t| t.get_column(name))
                    .map(|c| c.nominal_type.clone())
                    .unwrap_or_else(|| "text".to_string())
            };
            let frag = postrust_core::query::QueryBuilder::logic_sql(&tree, resolver)?;
            parts.push(shift_placeholders(
                frag.sql(),
                self.base + self.params.len(),
            ));
            self.params.extend(frag.params().iter().cloned());
        }

        Ok(Some(format!("({})", parts.join(" AND "))))
    }
}

/// Record the first column a logic tree names that the table does not have.
fn check_logic_columns(
    tree: &postrust_core::api_request::LogicTree,
    table: Option<&postrust_core::schema_cache::Table>,
    unknown: &mut Option<String>,
) {
    use postrust_core::api_request::LogicTree;

    match tree {
        LogicTree::Expr { children, .. } => {
            for child in children {
                check_logic_columns(child, table, unknown);
            }
        }
        LogicTree::Stmt(filter) => {
            if unknown.is_none()
                && table
                    .map(|t| t.get_column(&filter.field.name).is_none())
                    .unwrap_or(true)
            {
                *unknown = Some(filter.field.name.clone());
            }
        }
    }
}

/// One level of the embed tree.
#[derive(Default)]
struct EmbedLevel {
    /// `(response key, SQL expression)` for each relation at this level.
    expressions: Vec<(String, String)>,
    /// `EXISTS` predicates from `!inner` on these relations. They reference
    /// the alias of the level *above*, so the caller applies them.
    inner_joins: Vec<String>,
    /// `ORDER BY` expressions for terms naming an embedded resource.
    orders: Vec<String>,
    /// Each embed under the name the client used for it, with its
    /// expression. A spread's response key is internal, so this is what a
    /// filter naming the embed -- `?clients=is.null` -- is matched against.
    filterable: Vec<(String, String)>,
    /// For each spread, the keys it contributes to its parent.
    ///
    /// Needed because a spread that matched nothing still carries them --
    /// as nulls, or as empty arrays for a to-many -- and by then there is no
    /// row to read the names off.
    spread_columns: Vec<(String, Vec<String>)>,
    /// Whether this level had any relation to embed at all.
    ///
    /// Distinct from having produced an expression: `clients()` is a relation
    /// that deliberately contributes nothing, and the fallback path must not
    /// take an absent expression for a relation the single-query form could
    /// not handle.
    saw_relations: bool,
}

/// The keys a spread's selection contributes to the object above it.
///
/// `*` names every column of the related table, an alias renames the column it
/// precedes, and a nested spread contributes its own keys, having already been
/// flattened into this one.
fn spread_output_names(
    schema_cache: &postrust_core::SchemaCache,
    child_qi: &postrust_core::api_request::QualifiedIdentifier,
    select: &[postrust_core::api_request::SelectItem],
) -> Vec<String> {
    use postrust_core::api_request::SelectItem;

    let mut names = Vec::new();
    for item in select {
        match item {
            SelectItem::Field { field, .. } if field.name == "*" => {
                if let Some(table) = schema_cache.get_table(child_qi) {
                    names.extend(table.columns.keys().cloned());
                }
            }
            SelectItem::Field { field, alias, .. } => {
                names.push(alias.clone().unwrap_or_else(|| field.name.clone()));
            }
            SelectItem::Relation {
                relation, alias, ..
            } => names.push(alias.clone().unwrap_or_else(|| relation.clone())),
            SelectItem::SpreadRelation {
                relation, select, ..
            } => {
                if let Some(rel) = schema_cache
                    .find_relationship(child_qi, relation, None, &child_qi.schema)
                    .ok()
                    .flatten()
                {
                    let target = rel.foreign_table().clone();
                    names.extend(spread_output_names(schema_cache, &target, select));
                }
            }
        }
    }
    names
}

/// Dissolve spread embeds into the objects that contain them.
///
/// A spread arrives as an ordinary JSON column under a reserved name. This
/// walks the response and, wherever it finds one, moves its columns up into
/// the surrounding object and drops the name.
///
/// It recurses first, so a spread nested inside another is already flattened
/// by the time its parent is -- which is what carries a grandchild's columns
/// all the way up.
///
/// Spreading a to-many gives each column an array of that column's values
/// across the matched rows, rather than an array of objects.
fn flatten_spreads(
    value: &mut serde_json::Value,
    columns: &std::collections::HashMap<String, Vec<String>>,
) {
    match value {
        serde_json::Value::Array(items) => {
            for item in items {
                flatten_spreads(item, columns);
            }
        }
        serde_json::Value::Object(object) => {
            for (_, child) in object.iter_mut() {
                flatten_spreads(child, columns);
            }

            let spread_keys: Vec<String> = object
                .keys()
                .filter(|key| key.starts_with(SPREAD_KEY_PREFIX))
                .cloned()
                .collect();

            for key in spread_keys {
                let Some(spread) = object.remove(&key) else {
                    continue;
                };
                let names = columns.get(&key).cloned().unwrap_or_default();
                match spread {
                    serde_json::Value::Object(columns) => {
                        for (name, column) in columns {
                            object.insert(name, column);
                        }
                    }
                    // A to-many: one array per column rather than an array of
                    // objects, so the rows are transposed. With no rows the
                    // columns are still named, each holding an empty array.
                    serde_json::Value::Array(rows) => {
                        for name in names {
                            let column = rows
                                .iter()
                                .map(|row| {
                                    row.get(&name).cloned().unwrap_or(serde_json::Value::Null)
                                })
                                .collect();
                            object.insert(name, serde_json::Value::Array(column));
                        }
                    }
                    // The relationship matched nothing. The keys are still
                    // the client's to expect, so they are there, holding null.
                    _ => {
                        for name in names {
                            object.insert(name, serde_json::Value::Null);
                        }
                    }
                }
            }
        }
        _ => {}
    }
}

/// The relation an embed names, found by the name the response uses.
fn embedded_by_name<'a>(
    select: &'a [postrust_core::api_request::SelectItem],
    name: &str,
) -> Option<(&'a str, Option<&'a str>)> {
    use postrust_core::api_request::SelectItem;

    select.iter().find_map(|item| match item {
        SelectItem::Relation {
            relation,
            alias,
            hint,
            ..
        } if alias.as_deref().unwrap_or(relation) == name => {
            Some((relation.as_str(), hint.as_deref()))
        }
        // A spread has no alias -- nothing is being named -- so it answers to
        // the relation's own name.
        SelectItem::SpreadRelation { relation, hint, .. } if relation == name => {
            Some((relation.as_str(), hint.as_deref()))
        }
        _ => None,
    })
}

/// Build the `ORDER BY` expressions for terms naming an embedded resource.
///
/// `order=clients(name)` orders by a column the parent does not have, so each
/// term becomes a scalar subselect fetched per parent row -- the same shape as
/// the embed itself, reduced to one column.
fn build_embed_orders(
    schema_cache: &postrust_core::SchemaCache,
    parent_qi: &postrust_core::api_request::QualifiedIdentifier,
    order: &[postrust_core::plan::CoercibleOrderTerm],
    select: &[postrust_core::api_request::SelectItem],
    ctx: &mut EmbedFilters<'_>,
) -> Result<Vec<String>, postrust_core::Error> {
    let mut orders = Vec::new();

    for term in order.iter() {
        let Some(relation) = &term.relation else {
            continue;
        };

        // `order=` names a resource this request embedded, by the name the
        // response gives it -- so `pros:projects(*)` is ordered on as `pros`,
        // and the table it came from is what the relationship is looked up
        // by. A name the selection does not carry is not an embedded resource
        // here, whether or not the schema has a table by that name.
        let embedded = embedded_by_name(select, relation)
            .ok_or_else(|| postrust_core::Error::NotAnEmbeddedResource(relation.clone()))?;

        let rel = schema_cache
            .find_relationship(parent_qi, embedded.0, embedded.1, &parent_qi.schema)?
            .ok_or_else(|| {
                schema_cache.relationship_not_found(
                    parent_qi,
                    embedded.0,
                    embedded.1,
                    &parent_qi.schema,
                )
            })?;

        // A resource that yields many rows per parent has no single value to
        // order on, so there is nothing the request could mean.
        if !rel.is_to_one() {
            return Err(postrust_core::Error::RelatedOrderNotPossible {
                origin: parent_qi.name.clone(),
                relation: relation.clone(),
            });
        }

        let plan = postrust_core::embed::EmbedPlan::resolve(rel, schema_cache)?;

        let child_alias = ctx.next_alias();

        // The column belongs to the related table, so its type has to be read
        // from there. Resolved against the parent it would be unknown, and an
        // unknown type is one this process asks the database to render -- for
        // a `jsonb` column that would wrap it in `to_jsonb` and then reach
        // into the wrapper.
        let child_qi = postrust_core::api_request::QualifiedIdentifier::new(
            &plan.foreign_schema,
            &plan.foreign_table,
        );
        let mut field = term.field.clone();
        field.to_json = !field.json_path.is_empty()
            && !schema_cache
                .get_table(&child_qi)
                .and_then(|table| table.get_column(&field.name))
                .is_some_and(|column| matches!(column.data_type.as_str(), "json" | "jsonb"));

        let column =
            postrust_core::query::QueryBuilder::qualified_column_sql(Some(&child_alias), &field);
        let mut expression =
            plan.order_expression("src", PARENT_ROW_COLUMN_REF, &child_alias, &column);

        match term.direction {
            Some(postrust_core::api_request::OrderDirection::Desc) => expression.push_str(" DESC"),
            Some(postrust_core::api_request::OrderDirection::Asc) => expression.push_str(" ASC"),
            None => {}
        }
        match term.nulls {
            Some(postrust_core::api_request::OrderNulls::First) => {
                expression.push_str(" NULLS FIRST")
            }
            Some(postrust_core::api_request::OrderNulls::Last) => {
                expression.push_str(" NULLS LAST")
            }
            None => {}
        }

        orders.push(expression);
    }

    Ok(orders)
}

/// The relationship an `order=relation(column)` term names, ready to order by.
///
/// Resolving it is the same work whether the term sits at the top level or
/// inside an embed, and it is the same mistake to skip: a relation term
/// rendered as a bare column name orders by a column of the wrong table --
/// silently, where the parent happens to have a column of that name, and with
/// `column "cost" does not exist` where it does not.
fn related_order_target(
    schema_cache: &postrust_core::SchemaCache,
    parent_qi: &postrust_core::api_request::QualifiedIdentifier,
    select: &[postrust_core::api_request::SelectItem],
    relation: &str,
    ctx: &mut EmbedFilters<'_>,
) -> Result<
    (
        postrust_core::embed::EmbedPlan,
        String,
        postrust_core::api_request::QualifiedIdentifier,
    ),
    postrust_core::Error,
> {
    // `order=` names a resource this request embedded, by the name the
    // response gives it -- so `pros:projects(*)` is ordered on as `pros`, and
    // the table it came from is what the relationship is looked up by. A name
    // the selection does not carry is not an embedded resource here, whether
    // or not the schema has a table by that name.
    let embedded = embedded_by_name(select, relation)
        .ok_or_else(|| postrust_core::Error::NotAnEmbeddedResource(relation.to_string()))?;

    let rel = schema_cache
        .find_relationship(parent_qi, embedded.0, embedded.1, &parent_qi.schema)?
        .ok_or_else(|| {
            schema_cache.relationship_not_found(
                parent_qi,
                embedded.0,
                embedded.1,
                &parent_qi.schema,
            )
        })?;

    // A resource that yields many rows per parent has no single value to order
    // on, so there is nothing the request could mean.
    if !rel.is_to_one() {
        return Err(postrust_core::Error::RelatedOrderNotPossible {
            origin: parent_qi.name.clone(),
            relation: relation.to_string(),
        });
    }

    let plan = postrust_core::embed::EmbedPlan::resolve(rel, schema_cache)?;
    let child_alias = ctx.next_alias();
    let child_qi = postrust_core::api_request::QualifiedIdentifier::new(
        &plan.foreign_schema,
        &plan.foreign_table,
    );

    Ok((plan, child_alias, child_qi))
}

/// The `ORDER BY` an embedded resource was asked for, if any.
///
/// `clients.order=name.desc` orders the rows inside the embed, which is a
/// property of the child's own subselect rather than of the parent.
/// `tasks.order=projects(id).desc` orders them by a column of a table the
/// child embeds in turn -- so it is resolved against the child, and rendered
/// as the same correlated subselect the top level uses.
#[allow(clippy::too_many_arguments)] // each names one part of where the embed sits
fn embed_order_sql(
    schema_cache: &postrust_core::SchemaCache,
    child_qi: &postrust_core::api_request::QualifiedIdentifier,
    child_select: &[postrust_core::api_request::SelectItem],
    child_alias: &str,
    ctx: &mut EmbedFilters<'_>,
    path: &[String],
    names: &[String],
) -> Result<Option<String>, postrust_core::Error> {
    use postrust_core::api_request::{OrderDirection, OrderNulls, OrderTerm};

    let terms: Vec<OrderTerm> = ctx
        .orders
        .iter()
        .filter(|(p, _)| p.as_slice() == path || p.as_slice() == names)
        .flat_map(|(_, terms)| terms.iter().cloned())
        .collect();

    if terms.is_empty() {
        return Ok(None);
    }

    let mut rendered = Vec::with_capacity(terms.len());
    for term in &terms {
        let (field, direction, nulls) = match term {
            OrderTerm::Field {
                field,
                direction,
                nulls,
            }
            | OrderTerm::Relation {
                field,
                direction,
                nulls,
                ..
            } => (field, direction, nulls),
        };

        let mut sql = match term {
            OrderTerm::Field { .. } => postrust_sql::escape_ident(&field.name),
            OrderTerm::Relation { relation, .. } => {
                let (plan, grandchild_alias, grandchild_qi) =
                    related_order_target(schema_cache, child_qi, child_select, relation, ctx)?;

                // The column belongs to the related table, so its type is read
                // from there rather than from the child.
                let pg_type = schema_cache
                    .get_table(&grandchild_qi)
                    .and_then(|table| table.get_column(&field.name))
                    .map(|column| column.data_type.clone())
                    .unwrap_or_default();
                let coercible = postrust_core::plan::CoercibleField::from_field(field, &pg_type);
                let column = postrust_core::query::QueryBuilder::qualified_column_sql(
                    Some(&grandchild_alias),
                    &coercible,
                );

                // The child is a real table alias at this level, so its alias
                // is also the expression for its whole row.
                plan.order_expression(child_alias, child_alias, &grandchild_alias, &column)
            }
        };

        match direction {
            Some(OrderDirection::Desc) => sql.push_str(" DESC"),
            Some(OrderDirection::Asc) => sql.push_str(" ASC"),
            None => {}
        }
        match nulls {
            Some(OrderNulls::First) => sql.push_str(" NULLS FIRST"),
            Some(OrderNulls::Last) => sql.push_str(" NULLS LAST"),
            None => {}
        }

        rendered.push(sql);
    }

    Ok(Some(rendered.join(", ")))
}

#[allow(clippy::too_many_arguments)] // each names one part of where the embed sits
fn build_embed_expressions(
    schema_cache: &postrust_core::SchemaCache,
    parent_qi: &postrust_core::api_request::QualifiedIdentifier,
    parent_alias: &str,
    // The SQL expression yielding the parent's whole row, which a computed
    // relationship takes as its argument. At a nested level the parent is a
    // real table alias and so is the row; at the top level the parent is a
    // derived table, whose alias is a `record` rather than the table's own
    // composite type, so a column carrying the row is passed instead.
    parent_row: &str,
    select: &[postrust_core::api_request::SelectItem],
    ctx: &mut EmbedFilters<'_>,
    path: &[String],
    // The same path spelled with each relation's own name rather than the
    // alias the request gave it.
    names: &[String],
) -> Result<EmbedLevel, postrust_core::Error> {
    use postrust_core::api_request::SelectItem;

    let mut level = EmbedLevel::default();

    for item in select {
        // A spread is embedded exactly like a plain relation. The two differ
        // only in where the child's columns end up, and that is settled once
        // the rows are JSON -- so here the spread is marked by the name its
        // expression is given, and flattened afterwards.
        let (relation, alias, child_select, join_type, hint, is_spread) = match item {
            SelectItem::Relation {
                relation,
                alias,
                select,
                join_type,
                hint,
            } => (relation, alias.clone(), select, join_type, hint, false),
            SelectItem::SpreadRelation {
                relation,
                select,
                join_type,
                hint,
            } => (relation, None, select, join_type, hint, true),
            SelectItem::Field { .. } => continue,
        };

        level.saw_relations = true;

        let rel = schema_cache
            .find_relationship(parent_qi, relation, hint.as_deref(), &parent_qi.schema)?
            .ok_or_else(|| {
                schema_cache.relationship_not_found(
                    parent_qi,
                    relation,
                    hint.as_deref(),
                    &parent_qi.schema,
                )
            })?;
        let plan = postrust_core::embed::EmbedPlan::resolve(rel, schema_cache)?;

        // No row to pass: the caller said the parent's is not expressible
        // here, and a computed relationship is a function of it.
        if plan.function.is_some() && parent_row.is_empty() {
            continue;
        }

        let child_alias = ctx.next_alias();
        let child_qi = postrust_core::api_request::QualifiedIdentifier::new(
            &plan.foreign_schema,
            &plan.foreign_table,
        );

        // Filters are addressed by the name the client used, which may be
        // either the alias or the relation itself: `c:clients(*)&c.id=eq.1`
        // and `the_tasks:tasks(*)&tasks.id=eq.1` both name the same embed.
        let mut child_path = path.to_vec();
        child_path.push(alias.clone().unwrap_or_else(|| relation.clone()));
        let mut child_names = names.to_vec();
        child_names.push(relation.clone());
        let child_where = ctx.predicate_for(&child_path, &child_names, &child_qi, schema_cache)?;

        // Deeper relations first: they become part of this level's SELECT list.
        let child_row = postrust_sql::escape_ident(&child_alias);
        let nested = build_embed_expressions(
            schema_cache,
            &child_qi,
            &child_alias,
            &child_row,
            child_select,
            ctx,
            &child_path,
            &child_names,
        )?;

        // A nested `!inner` is written against this child's alias, which is
        // only in scope inside this child's own subselect -- so it narrows the
        // child here rather than travelling up to the top-level wrapper.
        let child_where = match (child_where, nested.inner_joins.is_empty()) {
            (existing, true) => existing,
            (Some(existing), false) => Some(format!(
                "{} AND {}",
                existing,
                nested.inner_joins.join(" AND ")
            )),
            (None, false) => Some(nested.inner_joins.join(" AND ")),
        };

        // `!inner` on this relation restricts the parent rows. It is emitted
        // against `parent_alias`, so it belongs to the caller's level; reusing
        // the same predicate string lets both places share placeholders.
        // The child keeps the same alias it has in the embed subselect. The
        // two are sibling scopes, never nested, so there is no collision --
        // and any nested predicate folded into `child_where` already names
        // that alias, so it has to be the one in scope here too.
        if matches!(join_type, Some(postrust_core::api_request::JoinType::Inner)) {
            level.inner_joins.push(plan.inner_join_predicate(
                parent_alias,
                parent_row,
                &child_alias,
                child_where.as_deref(),
            ));
        }

        level.spread_columns.extend(nested.spread_columns);
        let nested_expressions = nested.expressions;

        let child_table =
            schema_cache.get_table(&postrust_core::api_request::QualifiedIdentifier::new(
                &plan.foreign_schema,
                &plan.foreign_table,
            ));

        // The child's own columns. Empty means every column, which is what a
        // relation with no explicit selection asks for.
        let mut parts: Vec<String> = Vec::new();
        let mut project_everything = child_select.is_empty();
        for nested_item in child_select {
            match nested_item {
                SelectItem::Field { field, .. } if field.name == "*" => {
                    project_everything = true;
                }
                SelectItem::Field { field, alias, .. } => {
                    let column = postrust_sql::escape_ident(&field.name);
                    // A schema that declared how one of its domains is written
                    // on the wire meant it wherever the value appears, and an
                    // embedded row is the same value in a different place.
                    let rendered = child_table
                        .as_ref()
                        .and_then(|table| table.get_column(&field.name))
                        .and_then(|c| schema_cache.representation(c.representation_type(), "json"))
                        .map(|function| format!("{}({})", function, column))
                        .unwrap_or_else(|| column.clone());
                    let name = alias.as_deref().unwrap_or(&field.name);
                    parts.push(format!(
                        "{} AS {}",
                        rendered,
                        postrust_sql::escape_ident(name)
                    ));
                }
                // Handled as an expression, not a column.
                SelectItem::Relation { .. } | SelectItem::SpreadRelation { .. } => {}
            }
        }
        if project_everything {
            parts.clear();
            parts.push(format!("{}.*", postrust_sql::escape_ident(&child_alias)));
        }
        for (field_name, expression) in nested_expressions {
            parts.push(format!(
                "{} AS {}",
                expression,
                postrust_sql::escape_ident(&field_name)
            ));
        }

        let inner_select = parts.join(", ");
        let child_order = embed_order_sql(
            schema_cache,
            &child_qi,
            child_select,
            &child_alias,
            ctx,
            &child_path,
            &child_names,
        )?;
        let (child_limit, child_offset) = ctx.range_for(&child_path, &child_names);
        let expression = plan.embed_expression(
            parent_alias,
            parent_row,
            &child_alias,
            &inner_select,
            child_limit,
            child_offset,
            child_where.as_deref(),
            child_order.as_deref(),
            // Positionally, with the parent row and nothing else: this surface
            // has no session to supply and no place to write an argument.
            None,
            ctx.rendering,
        )?;

        if is_spread {
            let names = spread_output_names(schema_cache, &child_qi, child_select);
            if names.is_empty() {
                // `...clients()` spreads no columns, so there is nothing to
                // fetch and nothing to merge.
                continue;
            }
            level
                .spread_columns
                .push((format!("{}{}", SPREAD_KEY_PREFIX, child_alias), names));
        }

        let key = match is_spread {
            // A name nothing can collide with, since the object it names is
            // dissolved into its parent before anyone sees it.
            true => format!("{}{}", SPREAD_KEY_PREFIX, child_alias),
            false => alias.clone().unwrap_or_else(|| relation.clone()),
        };

        // Under the name the client used, whatever the response key is: a
        // filter naming the embed is written against that name.
        //
        // What it records is an existence test, not the expression. Asking
        // whether a to-many embed is null would never be true -- it renders as
        // `[]`, never as null -- and "did this match anything" is the question
        // either way.
        level.filterable.push((
            alias.clone().unwrap_or_else(|| relation.clone()),
            plan.inner_join_predicate(
                parent_alias,
                parent_row,
                &child_alias,
                child_where.as_deref(),
            ),
        ));

        // `clients()` asks for none of the related resource's columns. It is
        // written to narrow the parent -- with `!inner`, or with a filter on
        // the embed -- and the resource itself is not part of the answer, so
        // it gets no key at all rather than an empty object. The test is
        // recursive: an embed whose only content is such an embed contributes
        // nothing either.
        if !yields_columns(child_select) {
            continue;
        }

        level.expressions.push((key, expression));
    }

    Ok(level)
}

/// The media type a function's own return type declares, where the client
/// asked for it.
///
/// A function returning a domain named `text/plain` returns text, and the
/// value is the whole response. It is still only returned that way to a client
/// that asked: the same value serialised as JSON is a perfectly good answer to
/// a request that did not.
fn declared_media_type(
    db_plan: &DbActionPlan,
    api_request: &postrust_core::api_request::ApiRequest,
) -> Option<(String, String)> {
    let DbActionPlan::Call { call, .. } = db_plan else {
        return None;
    };
    let declared = call.media_type.as_deref()?;
    let base = call.media_base_type.clone().unwrap_or_default();
    // A function declaring `*/*` says it will render whatever was asked for,
    // so every request is one it answers -- and what it answers with is bytes,
    // which on the wire is called `application/octet-stream`.
    if declared == "*/*" {
        return Some(("application/octet-stream".to_string(), base));
    }
    api_request
        .accept_media_types
        .iter()
        .any(|accepted| accepted.content_type() == declared)
        .then(|| (declared.to_string(), base))
}

/// Whether a selection produces any keys in the response.
///
/// A relation contributes only what its own selection does, so `tasks()` and
/// `tasks(projects())` are alike empty, while `tasks(name)` is not.
fn yields_columns(select: &[postrust_core::api_request::SelectItem]) -> bool {
    use postrust_core::api_request::SelectItem;

    select.iter().any(|item| match item {
        SelectItem::Field { .. } => true,
        SelectItem::Relation { select, .. } | SelectItem::SpreadRelation { select, .. } => {
            yields_columns(select)
        }
    })
}

/// Attach the relations requested in `select` onto `rows`.
///
/// One query per relation per level: the parents' join keys are collected and
/// passed as a single array, so embedding across a page of parents does not
/// become a query per row. Recurses for nested embeds.
///
/// The schema cache is taken for each relation while its query is planned, and
/// let go before the query runs. A reload waits for every guard to go, and
/// every request after it waits for the reload: a guard held across a query
/// holds them all up for as long as the query takes.
async fn embed_relations<S: State, T: Tx>(
    state: &S,
    conn: &mut T,
    parent_qi: &postrust_core::api_request::QualifiedIdentifier,
    select: &[postrust_core::api_request::SelectItem],
    rows: &mut [serde_json::Value],
    max_rows: Option<i64>,
) -> Result<(), postrust_core::Error> {
    use postrust_core::api_request::SelectItem;

    if rows.is_empty() {
        return Ok(());
    }

    for item in select {
        // A spread embed is fetched exactly like a plain one -- the two
        // differ only in where the child's columns end up, which is
        // settled once the rows are in hand.
        let (relation, alias, nested, is_spread, hint) = match item {
            SelectItem::Relation {
                relation,
                alias,
                select: nested,
                hint,
                ..
            } => (relation, alias.clone(), nested, false, hint),
            SelectItem::SpreadRelation {
                relation,
                select: nested,
                hint,
                ..
            } => (relation, None, nested, true, hint),
            SelectItem::Field { .. } => continue,
        };

        let guard = state.schema_cache().await;
        let schema_cache: &postrust_core::SchemaCache = &guard;
        let rel = schema_cache
            .find_relationship(parent_qi, relation, hint.as_deref(), &parent_qi.schema)?
            .ok_or_else(|| {
                schema_cache.relationship_not_found(
                    parent_qi,
                    relation,
                    hint.as_deref(),
                    &parent_qi.schema,
                )
            })?;

        let plan = postrust_core::embed::EmbedPlan::resolve(rel, schema_cache)?;
        let keys = postrust_core::embed::parent_keys(rows, &plan.local_column);

        // Which of the child's columns the query needs to return.
        //
        // A column that was not asked for still costs a heap read, JSON
        // serialisation in PostgreSQL, socket bytes and a parse before
        // being discarded, so the projection is worked out before the query
        // rather than filtered afterwards.
        //
        // A nested embed joins on a column of the child row, so that column
        // has to survive the projection even when the client did not ask
        // for it. A spread relation falls back to every column: its shape
        // is decided further down and is not a plain projection.
        let child_qi = postrust_core::api_request::QualifiedIdentifier::new(
            &plan.foreign_schema,
            &plan.foreign_table,
        );

        let mut child_columns: Vec<String> = Vec::new();
        let mut project_everything = nested.is_empty();
        for nested_item in nested {
            match nested_item {
                SelectItem::Field { field, .. } => child_columns.push(field.name.clone()),
                SelectItem::Relation { relation, .. } => {
                    match schema_cache
                        .find_relationship(&child_qi, relation, None, &child_qi.schema)
                        .ok()
                        .flatten()
                    {
                        Some(nested_rel) => {
                            let nested_plan =
                                postrust_core::embed::EmbedPlan::resolve(nested_rel, schema_cache)?;
                            child_columns.push(nested_plan.local_column);
                        }
                        // Leave it to the recursive call to report the
                        // unknown relation, rather than failing here with
                        // less context.
                        None => project_everything = true,
                    }
                }
                SelectItem::SpreadRelation { .. } => project_everything = true,
            }
        }
        if project_everything {
            child_columns.clear();
        }
        drop(guard);

        let mut grouped = if keys.is_empty() {
            std::collections::HashMap::new()
        } else {
            let sql = plan.children_grouped_sql(max_rows, &child_columns)?;

            let keys = postrust_sql::SqlParam::Array(
                keys.into_iter().map(postrust_sql::SqlParam::Text).collect(),
            );
            let span = embed_span(relation, &child_qi);
            if state.config().otel_db_statement {
                span.record("db.query.text", sql.as_str());
            }
            let fetched = conn.fetch_values(&sql, &[keys]).instrument(span).await?;

            // The query returns the join key and a JSON array of that
            // key's children, so the values are read straight out of the
            // columns -- the typed row converter would wrap them under the
            // expression name.
            let pairs: Vec<(serde_json::Value, serde_json::Value)> = fetched
                .into_iter()
                .filter_map(|row| {
                    let mut columns = row.into_iter();
                    Some((columns.next()?, columns.next()?))
                })
                .collect();

            postrust_core::embed::group_from_aggregated(pairs)
        };

        // Deeper embeds, if any were asked for.
        //
        // A nested embed issues one query for every child row at this
        // level, so the rows have to be contiguous for it. They arrive
        // grouped, so they are flattened for the recursive call and put
        // back afterwards. When nothing deeper was requested this is all
        // skipped, which is the common case.
        let has_deeper_embed = nested.iter().any(|item| {
            matches!(
                item,
                SelectItem::Relation { .. } | SelectItem::SpreadRelation { .. }
            )
        });

        if has_deeper_embed {
            let mut order: Vec<(String, usize)> = Vec::with_capacity(grouped.len());
            let mut flat: Vec<serde_json::Value> = Vec::new();
            for (key, children) in grouped.drain() {
                order.push((key, children.len()));
                flat.extend(children);
            }

            // Boxed: a recursive `async fn` call has to be.
            Box::pin(embed_relations(
                state, &mut *conn, &child_qi, nested, &mut flat, max_rows,
            ))
            .await?;

            let mut rest = flat.into_iter();
            for (key, count) in order {
                grouped.insert(key, rest.by_ref().take(count).collect());
            }
        }

        // Return only the requested columns of the related resource. An
        // empty nested select means every column.
        //
        // The projection in the query covers the columns; this covers what
        // is left over, which is the join column when the client did not
        // ask for it.
        // A spread picks its own columns out when it merges them, and does
        // it by the child's own column names, so pruning here would only
        // get in the way.
        let requested: Option<std::collections::HashSet<String>> = if nested.is_empty() || is_spread
        {
            None
        } else {
            Some(
                nested
                    .iter()
                    .flat_map(|nested_item| match nested_item {
                        SelectItem::Field { field, alias, .. } => {
                            vec![alias.clone().unwrap_or_else(|| field.name.clone())]
                        }
                        SelectItem::Relation {
                            relation, alias, ..
                        } => vec![alias.clone().unwrap_or_else(|| relation.clone())],
                        // A spread one level down has already merged
                        // its columns into these rows, so they are
                        // part of what was asked for.
                        SelectItem::SpreadRelation { select, .. } => select
                            .iter()
                            .filter_map(|column| match column {
                                SelectItem::Field { field, .. } => Some(field.name.clone()),
                                _ => None,
                            })
                            .collect(),
                    })
                    .collect(),
            )
        };

        if let Some(requested) = requested {
            for group in grouped.values_mut() {
                for child in group.iter_mut() {
                    if let Some(object) = child.as_object_mut() {
                        object.retain(|key, _| requested.contains(key));
                    }
                }
            }
        }
        if is_spread {
            // The child rows carry the table's own column names, so a
            // column is read by its real name and written under its alias.
            let spread_columns: Vec<(String, String)> = nested
                .iter()
                .filter_map(|nested_item| match nested_item {
                    SelectItem::Field { field, alias, .. } => Some((
                        field.name.clone(),
                        alias.clone().unwrap_or_else(|| field.name.clone()),
                    )),
                    _ => None,
                })
                .collect();
            for row in rows.iter_mut() {
                postrust_core::embed::spread_into_parent(row, &plan, &grouped, &spread_columns);
            }
        } else {
            let field_name = alias.clone().unwrap_or_else(|| relation.clone());
            for row in rows.iter_mut() {
                postrust_core::embed::attach_to_parent(row, &field_name, &plan, &grouped);
            }
        }
    }

    Ok(())
}

/// The first row of a statement that always returns one, as `fetch_one`
/// would have read it.
fn first_row(
    rows: Vec<Vec<serde_json::Value>>,
) -> Result<Vec<serde_json::Value>, postrust_core::Error> {
    rows.into_iter().next().ok_or_else(|| {
        postrust_core::Error::Internal(
            "no rows returned by a query that expected to return at least one row".into(),
        )
    })
}

/// Build an HTTP response from our response type.
fn build_response(response: PgrstResponse) -> Response<Bytes> {
    let mut builder = Response::builder().status(response.status);

    for (key, value) in &response.headers {
        builder = builder.header(key, value);
    }

    // Every header has been checked by now, so this should not fail. If it
    // does, an empty 200 would tell the client its request succeeded with
    // nothing to say; a 500 and a log line at least say that something broke.
    builder.body(response.body).unwrap_or_else(|e| {
        error!("Could not build the response: {}", e);
        error_response(postrust_core::Error::Internal(e.to_string()), false)
    })
}

/// Build an error response.
///
/// In production mode (PGRST_DEBUG=false or unset), sensitive error details
/// are hidden to prevent information leakage.
/// Build the response for a failed request.
///
/// `verbatim_db_errors` passes a database failure through as PostgreSQL
/// reported it -- its SQLSTATE, message, detail and hint -- which is what
/// PostgREST does and what a client written against PostgREST branches on.
/// It is off unless compatibility mode is on, because those fields describe
/// the schema rather than the request: a constraint's name, a column that was
/// not selected, the text of a failing query. Everything else is reported the
/// same either way.
pub fn error_response(error: postrust_core::Error, verbatim_db_errors: bool) -> Response<Bytes> {
    let status = error.status_code();

    // Check if debug mode is enabled
    let debug_mode = std::env::var("PGRST_DEBUG")
        .map(|v| v == "true" || v == "1")
        .unwrap_or(false);

    // A function that raised `sqlstate 'PGRST'` supplied the whole response:
    // its status, its headers and its body. Nothing here is the API layer's to
    // decide, including whether to sanitise it -- the schema wrote it.
    if let postrust_core::Error::Database(db) = &error {
        match db.raised_response() {
            Some(Ok(raised)) => {
                let mut builder = Response::builder()
                    .status(status)
                    .header("content-type", "application/json");
                for (name, value) in &raised.headers {
                    builder = builder.header(name, value);
                }
                return builder
                    .body(Bytes::from(
                        serde_json::to_vec(&raised.body).unwrap_or_default(),
                    ))
                    .unwrap_or_else(|_| Response::new(Bytes::new()));
            }
            // The schema meant to write a response and wrote something else.
            // Reported as the fault it is, and reported the same way whether
            // or not database errors are passed through, since this one is
            // about the schema's own text rather than about the request.
            Some(Err(fault)) => {
                return error_response(
                    postrust_core::Error::RaiseNotUnderstood(fault),
                    verbatim_db_errors,
                )
            }
            None => {}
        }
    }

    let body = if debug_mode {
        // Full error details in debug mode
        serde_json::to_vec(&error.to_json()).unwrap_or_default()
    } else if let (true, postrust_core::Error::Database(db)) = (verbatim_db_errors, &error) {
        serde_json::to_vec(&serde_json::json!({
            "code": db.code,
            "message": db.message,
            "details": db.details,
            "hint": db.hint,
        }))
        .unwrap_or_default()
    } else {
        // Sanitized error in production
        let sanitized = serde_json::json!({
            "code": error.code(),
            "message": sanitize_error_message(&error),
            "details": error.details(),
            // A hint says how to correct the request. It is carried by the
            // error itself, so it never has to reach for anything the
            // sanitised message deliberately left out.
            "hint": error.hint(),
        });
        serde_json::to_vec(&sanitized).unwrap_or_default()
    };

    let mut builder = Response::builder()
        .status(status)
        .header("content-type", "application/json");

    // A 401 has to say what would satisfy it, or a client has no way to know
    // it should be sending a token at all. Where a token *was* sent and was
    // not accepted, the challenge says which -- `invalid_token` with the
    // reason, so a client can tell "your key is wrong" from "your clock is"
    // without parsing the body.
    if status == StatusCode::UNAUTHORIZED {
        let challenge = match &error {
            postrust_core::Error::InvalidJwt(_)
            | postrust_core::Error::NoSuitableJwtKey
            | postrust_core::Error::JwtClaim(_) => format!(
                "Bearer error=\"invalid_token\", error_description=\"{}\"",
                error
            ),
            _ => "Bearer".to_string(),
        };
        builder = builder.header("www-authenticate", challenge);
    }

    builder
        .body(Bytes::from(body))
        .unwrap_or_else(|_| Response::new(Bytes::new()))
}

/// Sanitize error messages for production.
fn sanitize_error_message(error: &postrust_core::Error) -> String {
    use postrust_core::Error;

    // A schema-cache miss names what the client asked for, and the client
    // already knows that -- it is what it sent. Saying so is what makes the
    // difference between "Resource not found" and a message the client can act
    // on, and it discloses nothing it did not supply. What stays sanitised is
    // the database's own account of a failure, which is about the schema
    // rather than about the request.
    match error {
        Error::TableNotFound { name, .. } | Error::NotFound(name) => {
            return format!("Could not find the table '{}' in the schema cache", name)
        }
        Error::FunctionNotFound {
            name,
            params,
            single_param,
            ..
        } => {
            // A body that is one value carries no argument names, so there is
            // no argument list to report -- naming the function is the whole
            // of what was looked for.
            let looked_for =
                match postrust_core::error::names_only_one_param(single_param.as_deref()) {
                    true => name.clone(),
                    false => postrust_core::error::function_signature(name, params),
                };
            return format!(
                "Could not find the function {} in the schema cache",
                looked_for
            );
        }
        Error::UnacceptableSchema { requested, .. } => {
            return format!("Invalid schema: {}", requested)
        }
        // Names only the view the client addressed and the operation it asked
        // for, which is what it has to change.
        Error::Database(db) if db.is_unwritable_view() => return db.message.clone(),
        // The candidate list is the whole point of the message: it names the
        // signatures that could not be told apart.
        Error::AmbiguousFunction { .. } | Error::AmbiguousRelationship { .. } => {
            // The candidates are the whole point of these messages: they name
            // what could not be told apart.
            return error.to_string();
        }
        _ => {}
    }

    (match error {
        Error::ColumnNotFound(_) => "Column not found",
        // The column is the client's own word and the relation is the one it
        // addressed, so naming both says nothing it did not send.
        Error::UnknownColumn { .. } => return error.to_string(),
        // Each of these says only what the request itself said: the resource
        // it named, the preference it sent, the shape it asked for. Repeating
        // that back tells the client nothing it did not already know, and is
        // the difference between a message it can act on and one it cannot.
        Error::RelationshipNotFound { .. }
        | Error::NotAnEmbeddedResource(_)
        | Error::RelatedOrderNotPossible { .. }
        | Error::InvalidPreferences(_)
        | Error::NotSingular { .. }
        | Error::InvalidRange(_)
        | Error::InvalidBody(_)
        | Error::UnsupportedMethod(_)
        | Error::InvalidRpcMethod(_)
        | Error::InvalidPutFilters
        | Error::PutLimitNotAllowed
        | Error::PutMatchingPk
        | Error::MaxAffectedExceeded(_)
        | Error::InvalidGucHeaders
        | Error::InvalidGucStatus
        | Error::InvalidMediaType(_)
        // Says only that the rows the request asked for carry no geometry,
        // which is about the request's own shape and is what the client has
        // to change.
        | Error::MissingGeometryColumn
        // Quotes the client's own URL back at it, and names the grammar the
        // API publishes. Neither is the schema's.
        | Error::UnparsableQuery { .. }
        // Says nothing about the schema beyond that one RAISE is malformed,
        // which is what the author has to fix.
        | Error::RaiseNotUnderstood(_)
        | Error::InvalidResourcePath => return error.to_string(),
        Error::InvalidPath(_) => "Invalid request path",
        // What the token failed on is the client's own token, so naming it
        // costs nothing and is the only way it can tell a bad signature from
        // an expired one.
        Error::InvalidJwt(_)
        | Error::NoSuitableJwtKey
        | Error::JwtClaim(_)
        | Error::MissingAuth => return error.to_string(),
        Error::InsufficientPermissions(_) => "Forbidden",

        // A fixed policy message that reveals nothing about the request or the
        // schema, so it is passed through verbatim -- and it is what PostgREST
        // says, which is what a client branching on it will be matching.
        Error::AggregatesNotAllowed => "Use of aggregate functions is not allowed",
        Error::InvalidHeader(_) | Error::InvalidQueryParam(_) => "Invalid request",
        Error::Database(_) => "Database error",
        Error::ConnectionPool(_) => "Service temporarily unavailable",
        Error::Internal(_) => "Internal server error",
        _ => "An error occurred",
    })
    .to_string()
}

#[cfg(test)]
mod tests {
    use super::{query_error_log, QueryErrorLog};
    use http::StatusCode;

    fn database(code: &str) -> postrust_core::db::DbError {
        reported(code, "message")
    }

    fn reported(code: &str, message: &str) -> postrust_core::db::DbError {
        postrust_core::db::DbError::database(
            "driver message",
            postrust_core::error::DatabaseError {
                code: code.into(),
                message: message.into(),
                details: None,
                hint: None,
                constraint: None,
                table: None,
                column: None,
            },
        )
    }

    #[test]
    fn what_the_schema_answers_on_purpose_is_a_refusal() {
        for (code, status) in [
            // A function answering "not found".
            ("PT404", StatusCode::NOT_FOUND),
            // A check constraint, a duplicate key.
            ("23514", StatusCode::BAD_REQUEST),
            ("23505", StatusCode::CONFLICT),
            // `id=eq.abc` on an integer column.
            ("22P02", StatusCode::BAD_REQUEST),
        ] {
            assert_eq!(
                query_error_log(&database(code)),
                QueryErrorLog::Refused(status),
                "{code}"
            );
        }
    }

    #[test]
    fn a_write_the_view_cannot_take_is_a_refusal() {
        assert_eq!(
            query_error_log(&reported("55000", "cannot update view \"j\"")),
            QueryErrorLog::Refused(StatusCode::METHOD_NOT_ALLOWED)
        );
        // The same SQLSTATE about anything else is the server's.
        assert_eq!(
            query_error_log(&reported("55000", "lock is not available")),
            QueryErrorLog::Fault
        );
    }

    #[test]
    fn faults_answered_with_a_4xx_are_still_faults() {
        for code in [
            "42601", // SQL the planner got wrong
            "42703", // likewise
            "42P01", // a dropped table the schema cache still lists
            "42501", // a missing grant
            "P0001", // a bare RAISE EXCEPTION
            "22012", // a function's arithmetic
        ] {
            assert!(
                database(code)
                    .as_database()
                    .unwrap()
                    .status_code()
                    .is_client_error(),
                "{code} is not a 4xx, so this case tests nothing"
            );
            assert_eq!(
                query_error_log(&database(code)),
                QueryErrorLog::Fault,
                "{code}"
            );
        }
    }

    #[test]
    fn server_errors_and_broken_connections_are_faults() {
        for code in ["XX000", "PT500"] {
            assert_eq!(
                query_error_log(&database(code)),
                QueryErrorLog::Fault,
                "{code}"
            );
        }
        let broken = postrust_core::db::DbError::connection("closed");
        assert_eq!(query_error_log(&broken), QueryErrorLog::Fault);
        assert_eq!(query_error_log(&broken.queued()), QueryErrorLog::Fault);
    }

    #[test]
    fn a_queued_refusal_stays_a_refusal_whatever_its_status() {
        for code in ["XX000", "PT403"] {
            assert_eq!(
                query_error_log(&database(code).queued()),
                QueryErrorLog::RefusedBefore
            );
        }
    }

    /// A data-modifying `WITH` may only appear at the top level, so anything
    /// that wraps the statement has to leave the clause where it is.
    #[test]
    fn a_leading_cte_is_lifted_off_the_statement() {
        let (clause, body) = super::split_leading_cte(
            "WITH pgrst_mutation_result AS (DELETE FROM t WHERE n = ')' RETURNING t.id) \
             SELECT \"id\" FROM \"pgrst_mutation_result\"",
        );

        assert_eq!(
            clause,
            "WITH pgrst_mutation_result AS (DELETE FROM t WHERE n = ')' RETURNING t.id) "
        );
        assert_eq!(body, "SELECT \"id\" FROM \"pgrst_mutation_result\"");
    }

    /// A plain read has no clause to lift.
    #[test]
    fn a_statement_without_a_cte_is_left_alone() {
        let sql = "SELECT \"id\" FROM \"test\".\"items\"";
        let (clause, body) = super::split_leading_cte(sql);

        assert!(clause.is_empty());
        assert_eq!(body, sql);
    }

    use super::*;
    use postrust_core::plan::CallParams;
    use postrust_core::QualifiedIdentifier;
    use serde_json::json;

    fn call_plan(name: &str, returns_set: bool) -> CallPlan {
        CallPlan {
            output_columns: Vec::new(),
            media_type: None,
            media_base_type: None,
            function: QualifiedIdentifier::new("public", name),
            params: CallParams::None,
            returns_scalar: !returns_set,
            return_type: None,
            returns_set,
            returns_composite: false,
            volatility: "Volatile".into(),
            param_types: Vec::new(),
            returns_void: false,
            variadic_params: Vec::new(),
        }
    }

    fn composite_call_plan(name: &str, returns_set: bool) -> CallPlan {
        CallPlan {
            returns_composite: true,
            returns_scalar: false,
            return_type: None,
            ..call_plan(name, returns_set)
        }
    }

    #[test]
    fn unwraps_json_return_to_bare_object() {
        // `SELECT * FROM sync(...)` on a json-returning function yields a single
        // column named after the function.
        let rows = vec![json!({"sync": {"ok": true, "count": 3}})];
        let (rows, singular) = unwrap_rpc_rows(rows, &call_plan("sync", false));
        assert!(singular, "non-set-returning function should be singular");
        assert_eq!(rows, vec![json!({"ok": true, "count": 3})]);
    }

    #[test]
    fn unwraps_scalar_return() {
        let rows = vec![json!({"add": 42})];
        let (rows, singular) = unwrap_rpc_rows(rows, &call_plan("add", false));
        assert!(singular);
        assert_eq!(rows, vec![json!(42)]);
    }

    #[test]
    fn unwraps_setof_scalar_to_array() {
        let rows = vec![json!({"gen": 1}), json!({"gen": 2})];
        let (rows, singular) = unwrap_rpc_rows(rows, &call_plan("gen", true));
        assert!(!singular, "set-returning function should not be singular");
        assert_eq!(rows, vec![json!(1), json!(2)]);
    }

    #[test]
    fn leaves_multi_column_rows_untouched() {
        // `RETURNS TABLE(...)` / composite set: rows already have real columns
        // and must not be un-nested.
        let rows = vec![json!({"id": 1, "name": "a"}), json!({"id": 2, "name": "b"})];
        let (out, singular) =
            unwrap_rpc_rows(rows.clone(), &composite_call_plan("list_users", true));
        assert!(!singular);
        assert_eq!(out, rows);
    }

    #[test]
    fn leaves_table_column_named_like_function_untouched() {
        // `CREATE FUNCTION foo() RETURNS TABLE(foo int)`: the single output
        // column legitimately shares the function's name. The composite
        // return-type metadata must prevent it from being mistaken for the
        // function-name wrapper.
        let rows = vec![json!({"foo": 1}), json!({"foo": 2})];
        let (out, singular) = unwrap_rpc_rows(rows.clone(), &composite_call_plan("foo", true));
        assert!(!singular);
        assert_eq!(out, rows);
    }

    #[test]
    fn single_composite_return_is_singular_but_not_unwrapped() {
        // A non-set function returning a row type expands to its columns;
        // nothing to un-nest, but the result still renders as a bare object.
        let rows = vec![json!({"id": 1, "name": "a"})];
        let (out, singular) =
            unwrap_rpc_rows(rows.clone(), &composite_call_plan("get_user", false));
        assert!(singular);
        assert_eq!(out, rows);
    }

    #[test]
    fn leaves_single_key_row_untouched_when_key_is_not_function_name() {
        // A single real column that happens to be the only column should not be
        // mistaken for the function-name wrapper.
        let rows = vec![json!({"id": 7})];
        let (out, _) = unwrap_rpc_rows(rows.clone(), &call_plan("get_thing", false));
        assert_eq!(out, rows);
    }
}