use crate::state::State;
use bytes::Bytes;
use http::{Response, StatusCode};
use http_body_util::BodyExt;
use postrust_auth::authenticate;
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, error};
use postrust_core::query::PARENT_ROW_COLUMN;
const SPREAD_KEY_PREFIX: &str = "pgrst_spread_";
const PARENT_ROW_COLUMN_REF: &str = "\"src\".\"pgrst_parent_row\"";
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;
match process_request(state, request).await {
Ok(response) => response,
Err(e) => error_response(e, verbatim_db_errors),
}
}
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>>,
{
let auth_header = request
.headers()
.get("authorization")
.and_then(|v| v.to_str().ok());
let auth_result = match state.jwt_cache() {
Some(cache) => cache.authenticate(auth_header, state.jwt_config()),
None => authenticate(auth_header, state.jwt_config()),
}
.map_err(jwt_error)?;
debug!("Authenticated as role: {}", auth_result.role);
let (parts, body) = request.into_parts();
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()))?;
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()))?;
let mut api_request = parse_request(
&http_request,
state.default_schema(),
state.schemas(),
state.config().db_max_rows,
)?;
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;
}
if api_request.query_params.columns.is_none() {
if let Some(payload) = &api_request.payload {
postrust_core::api_request::payload::validate_uniform_keys(payload)?;
}
}
if api_request.payload.is_none() && needs_payload(&api_request) {
return Err(postrust_core::Error::InvalidBody(
"Empty or invalid json".into(),
));
}
if let Some(plan) = requested_plan(&api_request) {
return Err(postrust_core::Error::InvalidMediaType(plan));
}
let schema_cache = state.schema_cache().await;
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);
}
let plan = create_action_plan(&api_request, &schema_cache)?;
drop(schema_cache);
let (result, tx) = execute_plan(
state,
&api_request,
&plan,
&auth_result,
&added_join_columns,
)
.await?;
let tx = match tx {
Some(tx) if is_read_only(&api_request) => {
tx.commit().await.map_err(postrust_core::Error::from)?;
None
}
tx => tx,
};
let response = format_response(&api_request, &result).map_err(|e| match e {
postrust_response::FormatError::MultipleRows | postrust_response::FormatError::NotFound => {
postrust_core::Error::NotSingular {
rows: result.rows.len().max(result.affected),
}
}
other => postrust_core::Error::Internal(other.to_string()),
})?;
if let Some(tx) = tx {
tx.commit().await.map_err(postrust_core::Error::from)?;
}
Ok(build_response(response))
}
fn requested_plan(api_request: &ApiRequest) -> Option<String> {
use postrust_core::api_request::MediaType;
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,
}
}
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,
}
}
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)
}
fn json_object<'a>(entries: impl IntoIterator<Item = (&'a String, &'a String)>) -> String {
serde_json::Value::Object(
entries
.into_iter()
.map(|(name, value)| (name.clone(), serde_json::Value::String(value.clone())))
.collect(),
)
.to_string()
}
const LOCATION_KEY_PREFIX: &str = "pgrst_location_";
const MEDIA_ROW_COUNT_COLUMN: &str = "pgrst_media_rows";
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,
..
})
)
}
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("&")))
}
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
}
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,
..
})
)
}
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,
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,
}
}
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) => {
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)
),
handler.table.is_some()
|| has_default_select(&api_request.query_params.select),
handler.base_type.clone(),
)
})
})
})
};
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(", "),
));
}
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();
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)
}
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());
}
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 {
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()
};
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(),
};
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
};
let query = postrust_core::query::build_query(
&ActionPlan::Db(db_plan.clone()),
Some(&auth.role),
)?;
if !query.has_main() {
return Ok((QueryResult::default(), None));
}
let (mut sql, params) = query.build_main();
let mut count_sql = match api_request.preferences.count {
Some(_) => unpaged_sql(db_plan, &auth.role)?,
None => None,
};
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,
rendering: match &media_handler {
Some(_) => postrust_core::EmbedRendering::Normalised,
None => postrust_core::EmbedRendering::Verbatim,
},
};
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;
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;
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,
)?;
}
let mut params = params;
params.extend(embed_filters.params);
if embed_level.saw_relations {
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)),
Some(&auth.role),
)?
.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));
}
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
);
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,
};
predicates.push(match negated {
true => exists.clone(),
false => format!("NOT {}", exists),
});
}
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(", "));
}
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));
}
}
}
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),
},
None => None,
}
} else {
None
};
if let Some((media_type, aggregate, over_row, _)) = &media_handler {
let argument = if *over_row {
format!(
"pgrst_media.{}",
postrust_sql::escape_ident(PARENT_ROW_COLUMN)
)
} else {
"pgrst_media".to_string()
};
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);
}
debug!("Executing SQL: {}", sql);
debug!("With {} parameters", params.len());
let mut tx = state
.db()
.begin()
.await
.map_err(|e| postrust_core::Error::ConnectionPool(e.to_string()))?;
let isolation = state
.config()
.role_settings
.get(&auth.role)
.and_then(|r| r.isolation_level.as_ref())
.unwrap_or(&state.config().db_tx_isolation);
if *isolation != postrust_core::IsolationLevel::ReadCommitted {
tx.queue(
&format!("SET TRANSACTION ISOLATION LEVEL {}", isolation.to_sql()),
&[],
);
}
if is_read_only(api_request) {
tx.queue("SET TRANSACTION READ ONLY", &[]);
}
{
let mut path = vec![api_request.schema.clone()];
path.extend(state.config().db_extra_search_path.iter().cloned());
let search_path = path
.iter()
.map(|s| postrust_sql::escape_ident(s))
.collect::<Vec<_>>()
.join(", ");
let mut settings: Vec<(String, String)> = Vec::new();
if let Some(role_settings) = state.config().role_settings.get(&auth.role) {
if let Some(timeout) = role_settings.statement_timeout {
settings.push(("statement_timeout".to_string(), timeout.to_string()));
}
}
settings.extend([
("search_path".to_string(), search_path),
("role".to_string(), auth.role.clone()),
("request.method".to_string(), api_request.method.clone()),
("request.path".to_string(), api_request.path.clone()),
(
"request.headers".to_string(),
json_object(&api_request.headers),
),
(
"request.cookies".to_string(),
json_object(&api_request.cookies),
),
]);
settings.extend(auth.claim_settings());
for (key, value) in &state.config().app_settings {
settings.push((format!("app.settings.{key}"), value.clone()));
}
if let Some(timezone) = &api_request.preferences.timezone {
settings.push(("timezone".to_string(), timezone.clone()));
}
let calls = (0..settings.len())
.map(|i| format!("set_config(${}, ${}, true)", i * 2 + 1, i * 2 + 2))
.collect::<Vec<_>>()
.join(", ");
let sql = format!("SELECT {}", calls);
let params: Vec<postrust_sql::SqlParam> = settings
.into_iter()
.flat_map(|(name, value)| {
[
postrust_sql::SqlParam::Text(name),
postrust_sql::SqlParam::Text(value),
]
})
.collect();
tx.queue(&sql, ¶ms);
}
if let Some(pre_request) = &state.config().db_pre_request {
let qualified = pre_request
.split('.')
.map(postrust_sql::escape_ident)
.collect::<Vec<_>>()
.join(".");
tx.queue(&format!("SELECT {qualified}()"), &[]);
}
let log = |e: postrust_core::db::DbError| {
match e.was_queued() && !e.is_connection() {
true => debug!("Refused before the query: {}", e),
false => error!("Query error: {}", e),
}
postrust_core::Error::from(e)
};
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, ¶ms, MEDIA_ROW_COUNT_COLUMN)
.await
.map_err(log)?;
let mut body = raw
.first
.map(|bytes| raw_column(bytes, &base_type))
.unwrap_or_default();
if state.config().compat_mode && base_type == "jsonb" {
body.insert(0, 0x01);
}
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,
¶ms,
api_request.max_rows,
)
.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),
));
}
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 { .. }
)
})
});
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();
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, ¶ms).await.map_err(log)?;
(rows, None, None)
} else if ends_with_the_query {
let (rows, settings) = tx
.fetch_json_then_and_commit(&sql, ¶ms, 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, ¶ms, RESPONSE_SETTINGS)
.await
.map_err(log)?;
(rows, Some(settings), Some(tx))
} else {
let rows = tx.fetch_json(&sql, ¶ms).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)?;
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))
}
};
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)?,
),
};
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, ¶ms, api_request.max_rows).await?
}
_ => None,
};
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);
}
}
}
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);
}
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)
};
let returns_void = matches!(
db_plan,
DbActionPlan::Call { call, .. } if call.returns_void
);
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));
}
}
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);
}
}
}
if is_upsert(api_request) && json_rows.len() != 1 {
return Err(postrust_core::Error::PutMatchingPk);
}
let location = build_location(api_request, &mut json_rows, &location_keys);
let omit_body = is_mutation(db_plan) && !wants_representation(api_request);
let affected = json_rows.len();
let rows = if omit_body { Vec::new() } else { json_rows };
let geojson_body = geojson_column
.as_ref()
.filter(|_| !omit_body)
.map(|column| {
(
"application/geo+json".to_string(),
feature_collection(&rows, column),
)
});
let offset = top_level_offset(api_request);
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,
)),
};
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 {
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()
};
Ok((result, tx))
}
ActionPlan::Info(info_plan) => {
use postrust_core::plan::InfoPlan;
let response_data = match info_plan {
InfoPlan::OpenApiSpec => {
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,
))
}
}
}
const RESPONSE_SETTINGS: &str = "SELECT current_setting('response.headers', true), \
current_setting('response.status', true)";
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(relation_allow(schema_cache.require_table(qi)?)),
Action::RoutineInfo { qi, .. } => Ok(routine_allow(schema_cache.get_routines(qi))),
_ => Ok("OPTIONS,GET,HEAD".to_string()),
}
}
fn relation_allow(table: &postrust_core::schema_cache::Table) -> String {
let mut methods = vec!["OPTIONS", "GET", "HEAD"];
if table.insertable {
methods.push("POST");
}
if table.insertable && table.updatable && !table.pk_cols.is_empty() {
methods.push("PUT");
}
if table.updatable {
methods.push("PATCH");
}
if table.deletable {
methods.push("DELETE");
}
methods.join(",")
}
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(),
}
}
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)
}
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()),
Action::Db(DbAction::RelationMut { qi, .. }) => Some(qi.clone()),
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,
}
}
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)?;
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}"),
})
}
}
}
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;
}
if let Ok(rounded) = format!("{float:.9}").parse::<f64>() {
if let Some(rounded) = serde_json::Number::from_f64(rounded) {
*number = rounded;
}
}
}
_ => {}
}
}
fn top_level_offset(api_request: &ApiRequest) -> i64 {
postrust_core::plan::resolve_top_level_range(api_request).offset
}
fn raw_column(bytes: Vec<u8>, base_type: &str) -> Vec<u8> {
if base_type != "bytea" {
return bytes;
}
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)
}
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,
}
}
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
)
}
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()
}
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"))
};
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())
}
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)
}
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,
Action::RelationInfo(_) | Action::RoutineInfo { .. } | Action::SchemaInfo => true,
}
}
fn unpaged_sql(
db_plan: &postrust_core::plan::DbActionPlan,
role: &str,
) -> 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), Some(role))?;
Ok(query.has_main().then(|| query.build_main().0))
}
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>,
) -> Result<Option<i64>, postrust_core::Error> {
use postrust_core::api_request::PreferCount;
match preference {
PreferCount::Exact => exact_count(tx, count_sql, params).await,
PreferCount::Planned => planned_count(tx, count_sql, params).await,
PreferCount::Estimated => match (exact_count(tx, count_sql, params).await?, max_rows) {
(Some(exact), Some(ceiling)) if exact <= ceiling => Ok(Some(exact)),
(Some(exact), _) => {
let planned = planned_count(tx, count_sql, params).await?;
Ok(Some(exact.max(planned.unwrap_or(exact))))
}
(None, _) => planned_count(tx, count_sql, params).await,
},
}
}
async fn exact_count<T: Tx>(
tx: &mut T,
count_sql: &str,
params: &[postrust_sql::SqlParam],
) -> Result<Option<i64>, postrust_core::Error> {
let sql = format!("SELECT count(*) FROM ({}) AS pgrst_count", count_sql);
let row = first_row(tx.fetch_values(&sql, params).await?)?;
Ok(row.first().and_then(serde_json::Value::as_i64))
}
async fn planned_count<T: Tx>(
tx: &mut T,
count_sql: &str,
params: &[postrust_sql::SqlParam],
) -> Result<Option<i64>, postrust_core::Error> {
let sql = format!("EXPLAIN (FORMAT JSON) {}", count_sql);
let plan = first_row(tx.fetch_values(&sql, params).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))
}
fn is_mutation(db_plan: &postrust_core::plan::DbActionPlan) -> bool {
matches!(
db_plan,
postrust_core::plan::DbActionPlan::MutateRead { .. }
)
}
fn wants_representation(api_request: &ApiRequest) -> bool {
matches!(
api_request.preferences.representation,
postrust_core::api_request::PreferRepresentation::Full
)
}
fn wants_location(api_request: &ApiRequest) -> bool {
matches!(
api_request.preferences.representation,
postrust_core::api_request::PreferRepresentation::HeadersOnly
)
}
fn mutation_status(
db_plan: &postrust_core::plan::DbActionPlan,
api_request: &ApiRequest,
created_a_row: bool,
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 {
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,
}
}
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,
},
})
}
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());
}
let selects_everything = select
.iter()
.any(|item| matches!(item, SelectItem::Field { field, .. } if field.name == "*"));
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() {
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)?;
if plan.function.is_some() {
if !added.iter().any(|c| c == PARENT_ROW_COLUMN) {
added.push(PARENT_ROW_COLUMN.to_string());
}
continue;
}
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)
}
struct EmbedFilters<'a> {
filters: &'a [(
postrust_core::api_request::EmbedPath,
postrust_core::api_request::Filter,
)],
params: Vec<postrust_sql::SqlParam>,
base: usize,
orders: &'a [(
postrust_core::api_request::EmbedPath,
Vec<postrust_core::api_request::OrderTerm>,
)],
ranges: &'a std::collections::HashMap<String, postrust_core::api_request::Range>,
logic: &'a [(
postrust_core::api_request::EmbedPath,
postrust_core::api_request::LogicTree,
)],
max_rows: Option<i64>,
alias_counter: usize,
rendering: postrust_core::EmbedRendering,
}
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<'_> {
fn next_alias(&mut self) -> String {
self.alias_counter += 1;
format!("e{}", self.alias_counter)
}
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 {
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 {
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 "))))
}
}
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());
}
}
}
}
#[derive(Default)]
struct EmbedLevel {
expressions: Vec<(String, String)>,
inner_joins: Vec<String>,
orders: Vec<String>,
filterable: Vec<(String, String)>,
spread_columns: Vec<(String, Vec<String>)>,
saw_relations: bool,
}
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
}
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);
}
}
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));
}
}
_ => {
for name in names {
object.insert(name, serde_json::Value::Null);
}
}
}
}
}
_ => {}
}
}
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()))
}
SelectItem::SpreadRelation { relation, hint, .. } if relation == name => {
Some((relation.as_str(), hint.as_deref()))
}
_ => None,
})
}
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;
};
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,
)
})?;
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();
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)
}
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,
> {
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,
)
})?;
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))
}
#[allow(clippy::too_many_arguments)] 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)?;
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,
);
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)] fn build_embed_expressions(
schema_cache: &postrust_core::SchemaCache,
parent_qi: &postrust_core::api_request::QualifiedIdentifier,
parent_alias: &str,
parent_row: &str,
select: &[postrust_core::api_request::SelectItem],
ctx: &mut EmbedFilters<'_>,
path: &[String],
names: &[String],
) -> Result<EmbedLevel, postrust_core::Error> {
use postrust_core::api_request::SelectItem;
let mut level = EmbedLevel::default();
for item in select {
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)?;
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,
);
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)?;
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,
)?;
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 ")),
};
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,
));
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);
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)
));
}
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(),
None,
ctx.rendering,
)?;
if is_spread {
let names = spread_output_names(schema_cache, &child_qi, child_select);
if names.is_empty() {
continue;
}
level
.spread_columns
.push((format!("{}{}", SPREAD_KEY_PREFIX, child_alias), names));
}
let key = match is_spread {
true => format!("{}{}", SPREAD_KEY_PREFIX, child_alias),
false => alias.clone().unwrap_or_else(|| relation.clone()),
};
level.filterable.push((
alias.clone().unwrap_or_else(|| relation.clone()),
plan.inner_join_predicate(
parent_alias,
parent_row,
&child_alias,
child_where.as_deref(),
),
));
if !yields_columns(child_select) {
continue;
}
level.expressions.push((key, expression));
}
Ok(level)
}
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();
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))
}
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)
}
})
}
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 {
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);
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);
}
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 fetched = conn.fetch_values(&sql, &[keys]).await?;
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)
};
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);
}
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());
}
}
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())],
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 {
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(())
}
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(),
)
})
}
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);
}
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)
})
}
pub fn error_response(error: postrust_core::Error, verbatim_db_errors: bool) -> Response<Bytes> {
let status = error.status_code();
let debug_mode = std::env::var("PGRST_DEBUG")
.map(|v| v == "true" || v == "1")
.unwrap_or(false);
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()));
}
Some(Err(fault)) => {
return error_response(
postrust_core::Error::RaiseNotUnderstood(fault),
verbatim_db_errors,
)
}
None => {}
}
}
let body = if 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 {
let sanitized = serde_json::json!({
"code": error.code(),
"message": sanitize_error_message(&error),
"details": error.details(),
"hint": error.hint(),
});
serde_json::to_vec(&sanitized).unwrap_or_default()
};
let mut builder = Response::builder()
.status(status)
.header("content-type", "application/json");
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()))
}
fn sanitize_error_message(error: &postrust_core::Error) -> String {
use postrust_core::Error;
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,
..
} => {
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)
}
Error::AmbiguousFunction { .. } | Error::AmbiguousRelationship { .. } => {
return error.to_string();
}
_ => {}
}
(match error {
Error::ColumnNotFound(_) => "Column not found",
Error::UnknownColumn { .. } => return error.to_string(),
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(_)
| Error::MissingGeometryColumn
| Error::UnparsableQuery { .. }
| Error::RaiseNotUnderstood(_)
| Error::InvalidResourcePath => return error.to_string(),
Error::InvalidPath(_) => "Invalid request path",
Error::InvalidJwt(_)
| Error::NoSuitableJwtKey
| Error::JwtClaim(_)
| Error::MissingAuth => return error.to_string(),
Error::InsufficientPermissions(_) => "Forbidden",
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 {
#[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\"");
}
#[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() {
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() {
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() {
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() {
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() {
let rows = vec![json!({"id": 7})];
let (out, _) = unwrap_rpc_rows(rows.clone(), &call_plan("get_thing", false));
assert_eq!(out, rows);
}
}