use crate::app::{self, handle_request};
use crate::custom;
use crate::state::AppState;
use anyhow::Result;
#[cfg(feature = "graphql")]
use axum::routing::{get, post};
use axum::{
extract::{DefaultBodyLimit, Request},
http::{HeaderValue, Method},
response::{IntoResponse, Json},
routing::{any, Route},
Router,
};
use sqlx::postgres::{PgConnectOptions, PgPoolOptions};
use std::convert::Infallible;
use std::sync::Arc;
use tower::{Layer, Service};
use tower_http::cors::{AllowOrigin, Any as CorsAny, CorsLayer};
use tracing::{error, info};
#[derive(Clone, Copy, Debug)]
#[non_exhaustive]
pub struct Host {
pub background_tasks: bool,
pub websockets: bool,
}
impl Host {
pub const SERVER: Host = Host {
background_tasks: true,
websockets: true,
};
pub const LAMBDA: Host = Host {
background_tasks: false,
websockets: false,
};
}
#[non_exhaustive]
pub struct App {
pub router: Router,
pub state: Arc<AppState>,
}
pub async fn build(config: postrust_core::AppConfig, host: Host) -> Result<App> {
Builder::new(config, host).build().await
}
#[must_use = "a Builder does nothing until `build` is called"]
pub struct Builder {
config: postrust_core::AppConfig,
host: Host,
nests: Vec<(String, Router<Arc<AppState>>)>,
layers: Vec<Wrap>,
}
type Wrap = Box<dyn FnOnce(Router) -> Router + Send>;
impl Builder {
pub fn new(config: postrust_core::AppConfig, host: Host) -> Self {
Builder {
config,
host,
nests: Vec::new(),
layers: Vec::new(),
}
}
pub fn nest(mut self, prefix: impl Into<String>, router: Router<Arc<AppState>>) -> Self {
self.nests.push((prefix.into(), router));
self
}
pub fn layer<L>(mut self, layer: L) -> Self
where
L: Layer<Route> + Clone + Send + Sync + 'static,
L::Service: Service<Request> + Clone + Send + Sync + 'static,
<L::Service as Service<Request>>::Response: IntoResponse + 'static,
<L::Service as Service<Request>>::Error: Into<Infallible> + 'static,
<L::Service as Service<Request>>::Future: Send + 'static,
{
self.layers
.push(Box::new(move |router: Router| router.layer(layer)));
self
}
pub async fn build(self) -> Result<App> {
let nests = mount_points(self.nests)?;
let jwt = jwt_config(&self.config)?;
let mut app = assemble(self.config, self.host, nests, jwt).await?;
for wrap in self.layers {
app.router = wrap(app.router);
}
Ok(app)
}
}
pub const RESERVED_PREFIXES: &[&str] = &[
"/api",
"/healthz",
"/v1/version",
"/v1/graphql",
"/v1alpha1/graphql",
"/admin",
"/_/health",
"/_/ready",
];
const CUSTOM_PREFIX: &str = "/_";
type Nests = Vec<(String, Router<Arc<AppState>>)>;
fn mount_points(nests: Nests) -> Result<Nests> {
let mut checked: Nests = Vec::with_capacity(nests.len());
for (given, router) in nests {
let taken: Vec<&str> = checked.iter().map(|(prefix, _)| prefix.as_str()).collect();
let prefix = check_prefix(&given, &taken)?;
checked.push((prefix, router));
}
Ok(checked)
}
fn check_prefix(given: &str, taken: &[&str]) -> Result<String> {
let trimmed = given.trim().trim_matches('/');
if trimmed.is_empty() {
anyhow::bail!("cannot nest routes at {given:?}: a prefix must name a path below /");
}
if trimmed
.split('/')
.any(|segment| segment.is_empty() || segment.contains(['{', '}', '*']))
{
anyhow::bail!(
"cannot nest routes at {given:?}: a prefix is literal path segments, \
without parameters, wildcards or empty segments"
);
}
let prefix = format!("/{trimmed}");
if prefix == CUSTOM_PREFIX {
anyhow::bail!(
"cannot nest routes at {given:?}: the server serves /_ itself; nest beneath it, \
at /_/<name>, instead"
);
}
if let Some(reserved) = RESERVED_PREFIXES
.iter()
.find(|reserved| overlaps(&prefix, reserved))
{
anyhow::bail!(
"cannot nest routes at {given:?}: it overlaps {reserved}, which the server serves itself"
);
}
if let Some(other) = taken.iter().find(|other| overlaps(&prefix, other)) {
anyhow::bail!("cannot nest routes at {given:?}: it overlaps {other}, nested earlier");
}
Ok(prefix)
}
fn warn_of_shadowing(nests: &Nests, schema_cache: &postrust_core::SchemaCache, schemas: &[String]) {
for (prefix, _) in nests {
let first = prefix
.trim_start_matches('/')
.split('/')
.next()
.unwrap_or_default();
if first == "rpc" {
tracing::warn!(
"compatibility mode: routes nested at {prefix} answer in place of the \
database functions REST serves at /rpc; they are still served at /api/rpc"
);
}
if let Some(table) = schema_cache
.tables
.keys()
.find(|qi| qi.name == first && schemas.contains(&qi.schema))
{
tracing::warn!(
"compatibility mode: routes nested at {prefix} answer in place of {}.{first}, \
which REST would serve at /{first}; it is still served at /api/{first}",
table.schema
);
}
}
}
fn beneath(path: &str, base: &str) -> bool {
path.strip_prefix(base)
.is_some_and(|rest| rest.is_empty() || rest.starts_with('/'))
}
fn overlaps(a: &str, b: &str) -> bool {
beneath(a, b) || beneath(b, a)
}
fn jwt_config(config: &postrust_core::AppConfig) -> Result<postrust_auth::JwtConfig> {
let jwt = postrust_engine::jwt_config(config)
.map_err(|e| anyhow::anyhow!("PGRST_JWT_SECRET: {e}"))?;
crate::jwks::attach(jwt, config).map_err(|e| anyhow::anyhow!("{e}"))
}
async fn assemble(
config: postrust_core::AppConfig,
host: Host,
nests: Nests,
jwt: postrust_auth::JwtConfig,
) -> Result<App> {
info!("Database: {}", mask_db_uri(&config.db_uri));
let connect_options: PgConnectOptions = config
.db_uri
.parse()
.map_err(|e| anyhow::anyhow!("invalid database URI: {e}"))?;
let connect_options = if config.db_prepared_statements {
connect_options
} else {
info!("Prepared statements disabled");
connect_options.statement_cache_capacity(0)
};
let pool = PgPoolOptions::new()
.max_connections(config.db_pool_size)
.acquire_timeout(std::time::Duration::from_secs(config.db_pool_timeout))
.connect_with(connect_options)
.await?;
info!("Connected to database");
let schema_cache = postrust_core::SchemaCache::load_with_search_path(
&pool,
&config.db_schemas,
&config.db_extra_search_path,
)
.await?;
info!("{}", schema_cache.summary());
if config.compat_mode {
warn_of_shadowing(&nests, &schema_cache, &config.db_schemas);
}
let state = Arc::new(AppState::with_jwt(pool, schema_cache, config.clone(), jwt));
#[cfg(feature = "otel")]
crate::otel::observe(&state);
crate::jwks::start(&state, host).await;
let (reloaded, schema_reloads) = tokio::sync::watch::channel(0u64);
if config.db_channel_enabled && host.background_tasks {
spawn_schema_reloader(state.clone(), config.db_channel.clone(), reloaded);
}
#[cfg(not(feature = "graphql"))]
drop(schema_reloads);
let api_router: Router<Arc<AppState>> = Router::new()
.route("/", any(handle_request))
.route("/{*path}", any(handle_request));
let mut app: Router<Arc<AppState>> = Router::new().nest("/api", api_router);
app = app
.route("/healthz", axum::routing::get(|| async { "OK" }))
.route(
"/v1/version",
axum::routing::get(|| async {
Json(serde_json::json!({ "version": env!("CARGO_PKG_VERSION") }))
}),
);
app = app.nest(CUSTOM_PREFIX, custom::custom_router());
info!("Custom routes enabled at /_");
for (prefix, router) in nests {
info!("Custom routes enabled at {}", prefix);
let router = match config.compat_mode {
true => router.layer(axum::middleware::map_response_with_state(
state.clone(),
crate::error::render_verbatim,
)),
false => router,
};
app = app.nest(&prefix, router);
}
#[cfg(feature = "graphql")]
{
use async_graphql_axum::GraphQLBatchRequest as GqlRequest;
use axum::extract::State as AxumState;
use axum::http::HeaderMap;
use postrust_graphql::handler::GraphQLState;
use postrust_graphql::schema::SchemaConfig;
#[cfg(feature = "admin-ui")]
{
info!("Admin UI enabled at /admin");
app = app.nest("/admin", crate::admin::admin_router());
}
let schema_cache_snapshot = state.schema_cache.read().await.clone();
let schema_cache_arc = Arc::new(schema_cache_snapshot);
let names_var = ["PGRST_GRAPHQL_METADATA", "PGRST_GRAPHQL_NAMES"]
.into_iter()
.find_map(|name| std::env::var(name).ok().map(|value| (name, value)));
let graphql_names = match names_var {
Some((var, value)) => match postrust_graphql::names::NameOverrides::parse(&value) {
Ok(names) => {
if !names.is_empty() {
info!("GraphQL names given for {} tables", names.len());
}
if names.placed_functions() > 0 {
info!(
"GraphQL roots given for {} functions",
names.placed_functions()
);
}
if names.has_permissions() {
info!(
"GraphQL permissions given for {} roles, {} grants",
names.roles().len(),
names.granted()
);
}
names
}
Err(e) => {
tracing::error!("{}: {}", var, e);
return Err(anyhow::anyhow!("{}: {}", var, e));
}
},
None => postrust_graphql::names::NameOverrides::default(),
};
let graphql_enum_values = read_enum_values(
&state.pool,
&*state.schema_cache.read().await,
&graphql_names,
)
.await;
let subscription_refresh_seconds = std::env::var("PGRST_SUBSCRIPTION_REFRESH")
.ok()
.and_then(|value| value.trim().parse::<u64>().ok())
.unwrap_or(30);
let mut graphql_config = SchemaConfig::default();
graphql_config.enable_subscriptions = host.websockets;
graphql_config.subscription_refresh_seconds = subscription_refresh_seconds;
graphql_config.max_rows = config.db_max_rows;
graphql_config.enable_federation = config.graphql_federation;
graphql_config.type_prefix = config.graphql_type_prefix.clone();
graphql_config.exposed_schemas = config.db_schemas.clone();
graphql_config.enum_values = graphql_enum_values;
graphql_config.names = graphql_names;
let graphql_state = match GraphQLState::new(
state.pool.clone(),
schema_cache_arc.clone(),
graphql_config.clone(),
) {
Ok(graphql_state) => {
if !host.websockets {
info!("GraphQL subscriptions disabled: this host cannot hold a WebSocket");
} else if let Err(e) = graphql_state.init_subscriptions().await {
tracing::warn!("Failed to initialize subscription broker: {}. Subscriptions may not work until triggers are created.", e);
} else {
info!("GraphQL subscriptions enabled");
}
info!("GraphQL endpoint enabled at /api/graphql");
Some(Arc::new(graphql_state))
}
Err(e) => {
tracing::error!(
"GraphQL schema could not be built, serving without it: {}",
e
);
None
}
};
{
let current = Arc::new(std::sync::RwLock::new(graphql_state));
spawn_graphql_rebuilder(
state.clone(),
current.clone(),
graphql_config,
schema_reloads,
);
#[derive(Clone)]
struct GraphQLAppState {
gql_state: Arc<std::sync::RwLock<Option<Arc<GraphQLState>>>>,
rest: Arc<AppState>,
hasura_auth: postrust_auth::HasuraAuthConfig,
config: Arc<postrust_core::AppConfig>,
}
impl GraphQLAppState {
fn gql(&self) -> Option<Arc<GraphQLState>> {
self.gql_state
.read()
.unwrap_or_else(|e| e.into_inner())
.clone()
}
}
let graphql_app_state = GraphQLAppState {
gql_state: current,
rest: state.clone(),
hasura_auth: postrust_auth::HasuraAuthConfig {
admin_secret: config.hasura_admin_secret.clone(),
unauthorized_role: config.hasura_unauthorized_role.clone(),
},
config: Arc::new(state.config.clone()),
};
async fn handle_graphql(
AxumState(app_state): AxumState<GraphQLAppState>,
axum::extract::OriginalUri(uri): axum::extract::OriginalUri,
headers: HeaderMap,
req: GqlRequest,
) -> (axum::http::StatusCode, Json<serde_json::Value>) {
let endpoint = match uri.path().starts_with("/v1alpha1/") {
true => postrust_graphql::hasura::Endpoint::Legacy,
false => postrust_graphql::hasura::Endpoint::Current,
};
match req.0 {
async_graphql::BatchRequest::Single(request) => {
let body = one_graphql(app_state, headers, request).await;
let status = axum::http::StatusCode::from_u16(
postrust_graphql::hasura::status_for(&body, endpoint),
)
.unwrap_or(axum::http::StatusCode::OK);
(status, Json(body))
}
async_graphql::BatchRequest::Batch(requests) => {
let mut answers = Vec::with_capacity(requests.len());
for request in requests {
answers.push(
one_graphql(app_state.clone(), headers.clone(), request).await,
);
}
(
axum::http::StatusCode::OK,
Json(serde_json::Value::Array(answers)),
)
}
}
}
#[derive(Clone)]
struct GraphQLIdentity {
auth_result: postrust_auth::AuthResult,
hasura_role: Option<String>,
session: std::collections::HashMap<String, String>,
elevated: bool,
headers: HeaderMap,
}
struct PreparedOperation {
schema: async_graphql::dynamic::Schema,
request: async_graphql::Request,
write: postrust_graphql::context::SharedWrite,
}
async fn ensure_keys(app_state: &GraphQLAppState, headers: &HeaderMap) {
let auth_header = headers.get("authorization").and_then(|v| v.to_str().ok());
postrust_auth::ensure_keys(auth_header, &app_state.rest.jwt_config).await;
}
fn graphql_identity(
app_state: &GraphQLAppState,
headers: HeaderMap,
) -> Result<GraphQLIdentity, serde_json::Value> {
let auth_header = headers.get("authorization").and_then(|v| v.to_str().ok());
let jwt_config = &app_state.rest.jwt_config;
let token = match &app_state.rest.jwt_cache {
Some(cache) => cache.authenticate(auth_header, jwt_config),
None => postrust_auth::authenticate(auth_header, jwt_config),
};
let token_verified = auth_header.is_some() && token.is_ok();
let token_refusal = match (&token, auth_header) {
(Ok(_) | Err(postrust_auth::JwtError::SecretMissing), _) | (Err(_), None) => {
None
}
(Err(postrust_auth::JwtError::NoIdentity), Some(_)) => Some(
postrust_graphql::hasura::missing_role_claim(&jwt_config.role_claim_key),
),
(Err(e), Some(_)) => {
Some(postrust_graphql::hasura::invalid_jwt(&e.to_string()))
}
};
let auth_result = match token {
Ok(auth) => auth,
Err(e) => {
tracing::debug!("GraphQL auth failed: {}, using anon role", e);
postrust_auth::AuthResult {
role: jwt_config
.anon_role
.clone()
.unwrap_or_else(|| "anon".to_string()),
claims: std::collections::HashMap::new(),
}
}
};
tracing::debug!(
"GraphQL request authenticated as role: {}",
auth_result.role
);
let pairs: Vec<(&str, &str)> = headers
.iter()
.filter_map(|(name, value)| {
value.to_str().ok().map(|value| (name.as_str(), value))
})
.collect();
let (hasura_role, session, elevated) = match (
postrust_auth::hasura::from_admin_secret(&app_state.hasura_auth, &pairs[..]),
token_refusal,
) {
(postrust_auth::SecretOutcome::Accepted(identity), _) => {
(Some(identity.role), identity.session, identity.elevated)
}
(postrust_auth::SecretOutcome::Rejected, _) => {
return Err(postrust_graphql::hasura::denied(
postrust_auth::hasura::SECRET_INVALID,
));
}
(_, Some(refusal)) => return Err(refusal),
_ if token_verified => {
match postrust_auth::hasura::role_for_token(&auth_result.claims, &pairs[..])
{
postrust_auth::hasura::TokenRole::Is(role) => (
role,
postrust_auth::hasura::session_from_claims(&auth_result.claims),
false,
),
postrust_auth::hasura::TokenRole::NotAllowed => {
return Err(postrust_graphql::hasura::denied(
postrust_auth::hasura::ROLE_NOT_ALLOWED,
));
}
}
}
_ if app_state.hasura_auth.is_configured() => {
match postrust_auth::hasura::unauthenticated(&app_state.hasura_auth) {
Some(identity) => (Some(identity.role), identity.session, false),
None => {
return Err(postrust_graphql::hasura::denied(
postrust_auth::hasura::SECRET_MISSING,
));
}
}
}
_ => (
postrust_auth::hasura::role_from_claims(&auth_result.claims),
postrust_auth::hasura::session_from_claims(&auth_result.claims),
false,
),
};
Ok(GraphQLIdentity {
auth_result,
hasura_role,
session,
elevated,
headers,
})
}
fn prepare_operation(
app_state: &GraphQLAppState,
identity: &GraphQLIdentity,
req: async_graphql::Request,
) -> Result<PreparedOperation, serde_json::Value> {
let GraphQLIdentity {
auth_result,
hasura_role,
session,
elevated,
headers,
} = identity.clone();
let Some(gql) = app_state.gql() else {
return Err(postrust_graphql::hasura::unavailable(
"the GraphQL schema could not be built; see the server log",
));
};
let pairs: Vec<(&str, &str)> = headers
.iter()
.filter_map(|(name, value)| {
value.to_str().ok().map(|value| (name.as_str(), value))
})
.collect();
if postrust_graphql::hasura::persisted_query(&req) {
return Err(postrust_graphql::hasura::not_supported(
"PersistedQueryNotSupported",
));
}
if let Some(message) = postrust_auth::hasura::unreadable_boolean_header(&pairs[..])
{
return Err(postrust_graphql::hasura::malformed(&message));
}
let Some(schema) = gql.schema_for(
hasura_role.as_deref(),
postrust_auth::hasura::backend_only_requested(&pairs[..], elevated),
) else {
return Err(postrust_graphql::hasura::denied(&format!(
"role \"{}\" is not defined in the permissions",
hasura_role.as_deref().unwrap_or_default()
)));
};
let schema_cache_ref = postrust_core::schema_cache::SchemaCacheRef::from_static(
(*gql.schema_cache).clone(),
);
let write: postrust_graphql::context::SharedWrite = Arc::default();
let gql_ctx = postrust_graphql::context::GraphQLContext::new(
gql.pool.clone(),
schema_cache_ref,
auth_result,
)
.with_session(session)
.with_identity(hasura_role.clone(), elevated)
.with_write(Arc::clone(&write))
.with_config(Arc::clone(&app_state.config));
let introspection_disabled_for = hasura_role
.as_deref()
.filter(|_| !elevated)
.filter(|role| gql.config.names.introspection_disabled(role));
let prepared = postrust_graphql::hasura::prepare(
Some(schema),
req,
introspection_disabled_for,
);
let request = match prepared {
Ok(request) => request,
Err((_, errors)) => {
let mut response = async_graphql::Response::new(async_graphql::Value::Null);
response.errors = errors;
return Err(postrust_graphql::hasura::envelope(response));
}
};
let request = request
.data(gql_ctx)
.data(gql.pool.clone())
.data(Arc::clone(&gql.broker));
Ok(PreparedOperation {
schema: schema.clone(),
request,
write,
})
}
async fn settle_write(
write: &postrust_graphql::context::SharedWrite,
response: &mut async_graphql::Response,
) {
if let Some(tx) = write.lock().await.take() {
let settled = if response.errors.is_empty() {
tx.commit().await
} else {
tx.rollback().await
};
if let Err(e) = settled {
tracing::error!("GraphQL mutation could not be settled: {}", e);
response
.errors
.push(async_graphql::ServerError::new(e.to_string(), None));
response.data = async_graphql::Value::Null;
}
}
}
async fn one_graphql(
app_state: GraphQLAppState,
headers: HeaderMap,
req: async_graphql::Request,
) -> serde_json::Value {
ensure_keys(&app_state, &headers).await;
let identity = match graphql_identity(&app_state, headers) {
Ok(identity) => identity,
Err(refusal) => return refusal,
};
let PreparedOperation {
schema,
request,
write,
} = match prepare_operation(&app_state, &identity, req) {
Ok(operation) => operation,
Err(refusal) => return refusal,
};
use tracing::Instrument as _;
let span = tracing::debug_span!(
"graphql",
"graphql.operation.name" = request.operation_name.as_deref(),
);
let mut response = schema.execute(request).instrument(span).await;
settle_write(&write, &mut response).await;
postrust_graphql::hasura::envelope(response)
}
fn hasura_response(body: serde_json::Value) -> async_graphql::Response {
let data = body
.get("data")
.cloned()
.and_then(|data| async_graphql::Value::from_json(data).ok())
.unwrap_or(async_graphql::Value::Null);
let mut response = async_graphql::Response::new(data);
for error in body
.get("errors")
.and_then(serde_json::Value::as_array)
.into_iter()
.flatten()
{
let message = error
.get("message")
.and_then(serde_json::Value::as_str)
.unwrap_or_default();
let mut server_error = async_graphql::ServerError::new(message, None);
if let Some(fields) = error
.get("extensions")
.and_then(serde_json::Value::as_object)
{
let mut extensions = async_graphql::ErrorExtensionValues::default();
for (name, value) in fields {
if let Ok(value) = async_graphql::Value::from_json(value.clone()) {
extensions.set(name, value);
}
}
server_error.extensions = Some(extensions);
}
response.errors.push(server_error);
}
response
}
fn connection_headers(
mut headers: HeaderMap,
payload: &serde_json::Value,
) -> HeaderMap {
let nested = payload
.get("headers")
.and_then(serde_json::Value::as_object);
for fields in [payload.as_object(), nested].into_iter().flatten() {
for (name, value) in fields {
let Some(value) = value.as_str() else {
continue;
};
if let (Ok(name), Ok(value)) = (
axum::http::HeaderName::from_bytes(name.as_bytes()),
axum::http::HeaderValue::from_str(value),
) {
headers.insert(name, value);
}
}
}
headers
}
#[derive(Clone)]
struct WebSocketExecutor(GraphQLAppState);
impl async_graphql::Executor for WebSocketExecutor {
async fn execute(
&self,
request: async_graphql::Request,
) -> async_graphql::Response {
use async_graphql::futures_util::StreamExt;
self.execute_stream(request, None)
.next()
.await
.unwrap_or_else(|| async_graphql::Response::new(async_graphql::Value::Null))
}
fn execute_stream(
&self,
request: async_graphql::Request,
session_data: Option<Arc<async_graphql::Data>>,
) -> async_graphql::futures_util::stream::BoxStream<'static, async_graphql::Response>
{
use async_graphql::futures_util::{stream, StreamExt};
let identity = session_data
.as_deref()
.and_then(|data| data.get(&std::any::TypeId::of::<GraphQLIdentity>()))
.and_then(|identity| identity.downcast_ref::<GraphQLIdentity>())
.cloned();
let Some(identity) = identity else {
let refusal =
postrust_graphql::hasura::denied("the connection was not initialised");
return stream::once(async move { hasura_response(refusal) }).boxed();
};
match prepare_operation(&self.0, &identity, request) {
Err(refusal) => {
stream::once(async move { hasura_response(refusal) }).boxed()
}
Ok(PreparedOperation {
schema,
request,
write,
}) => {
async_graphql::Executor::execute_stream(&schema, request, session_data)
.then(move |mut response| {
let write = Arc::clone(&write);
async move {
settle_write(&write, &mut response).await;
hasura_response(postrust_graphql::hasura::envelope(
response,
))
}
})
.boxed()
}
}
}
}
async fn graphql_ws(
AxumState(app_state): AxumState<GraphQLAppState>,
protocol: async_graphql_axum::GraphQLProtocol,
headers: HeaderMap,
ws: axum::extract::ws::WebSocketUpgrade,
) -> axum::response::Response {
ws.protocols(async_graphql::http::ALL_WEBSOCKET_PROTOCOLS)
.on_upgrade(move |socket| async move {
let init_state = app_state.clone();
async_graphql_axum::GraphQLWebSocket::new(
socket,
WebSocketExecutor(app_state),
protocol,
)
.on_connection_init(move |payload| async move {
let headers = connection_headers(headers, &payload);
ensure_keys(&init_state, &headers).await;
match graphql_identity(&init_state, headers) {
Ok(identity) => {
let mut data = async_graphql::Data::default();
data.insert(identity);
Ok(data)
}
Err(refusal) => Err(async_graphql::Error::new(
refusal
.pointer("/errors/0/message")
.and_then(serde_json::Value::as_str)
.unwrap_or("connection refused"),
)),
}
})
.serve()
.await
})
}
let graphql_router = Router::new()
.route("/", post(handle_graphql))
.route("/", get(postrust_graphql::handler::graphql_playground))
.with_state(graphql_app_state.clone());
let ws_router = Router::new()
.route("/ws", get(graphql_ws))
.with_state(graphql_app_state);
let graphql_app = match host.websockets {
true => graphql_router.merge(ws_router),
false => graphql_router,
};
app = app
.nest("/v1/graphql", graphql_app.clone())
.nest("/v1alpha1/graphql", graphql_app.clone())
.nest("/api/graphql", graphql_app);
}
}
if config.compat_mode {
app = app.fallback(handle_request);
info!("PostgREST compatibility mode enabled: REST surface also served at /");
if !cfg!(feature = "compat-key-order") {
tracing::warn!(
"compatibility mode: object keys will be returned in alphabetical order, \
not in select order as PostgREST returns them. Build with \
--features compat-key-order to match, at a cost of up to 15% throughput \
on wide rows."
);
}
}
if !config.compat_mode {
app = app.route(
"/",
axum::routing::get(|| async {
Json(serde_json::json!({
"name": "postrust",
"version": env!("CARGO_PKG_VERSION"),
"api": "/api",
"custom": "/_",
"health": "/_/health",
"admin": "/admin",
"docs": "/admin/swagger"
}))
}),
);
}
let allow_origin = if config.server_cors_origins.is_empty() {
AllowOrigin::any()
} else {
let origins: Vec<HeaderValue> = config
.server_cors_origins
.iter()
.filter_map(|o| match HeaderValue::from_str(o) {
Ok(v) => Some(v),
Err(_) => {
tracing::warn!("Ignoring CORS origin {:?}: not a valid header value", o);
None
}
})
.collect();
info!("CORS restricted to {} origin(s)", origins.len());
AllowOrigin::list(origins)
};
let app = app
.layer(
CorsLayer::new()
.allow_origin(allow_origin)
.allow_methods([
Method::GET,
Method::POST,
Method::PUT,
Method::PATCH,
Method::DELETE,
Method::OPTIONS,
Method::HEAD,
])
.allow_headers(CorsAny)
.expose_headers(CorsAny),
)
.layer(DefaultBodyLimit::max(config.max_body_size))
.layer(axum::middleware::from_fn_with_state(
state.clone(),
app::options_allow,
))
.layer(crate::telemetry::RequestTraceLayer)
.with_state(state.clone());
Ok(App { router: app, state })
}
fn spawn_schema_reloader(
state: Arc<AppState>,
channel: String,
reloaded: tokio::sync::watch::Sender<u64>,
) {
const RETRY: std::time::Duration = std::time::Duration::from_secs(5);
tokio::spawn(async move {
loop {
let mut listener = match sqlx::postgres::PgListener::connect_with(&state.pool).await {
Ok(listener) => listener,
Err(e) => {
error!("Schema reloader could not connect: {}; retrying", e);
tokio::time::sleep(RETRY).await;
continue;
}
};
if let Err(e) = listener.listen(&channel).await {
error!("Schema reloader could not LISTEN on {}: {}", channel, e);
tokio::time::sleep(RETRY).await;
continue;
}
info!("Listening on channel {} for schema reloads", channel);
loop {
match listener.recv().await {
Ok(notification) => {
info!("Schema reload requested: {:?}", notification.payload());
match state.reload_schema().await {
Ok(()) => {
info!(
"Schema cache reloaded: {}",
state.schema_cache().await.summary()
);
reloaded.send_modify(|n| *n += 1);
#[cfg(feature = "otel")]
crate::otel::schema_reloaded(true);
}
Err(e) => {
error!("Schema cache reload failed: {}", e);
#[cfg(feature = "otel")]
crate::otel::schema_reloaded(false);
}
}
}
Err(e) => {
error!("Schema reloader lost its connection: {}; reconnecting", e);
break;
}
}
}
}
});
}
#[cfg(feature = "graphql")]
fn spawn_graphql_rebuilder(
state: Arc<AppState>,
current: Arc<std::sync::RwLock<Option<Arc<postrust_graphql::handler::GraphQLState>>>>,
mut config: postrust_graphql::schema::SchemaConfig,
mut reloads: tokio::sync::watch::Receiver<u64>,
) {
use postrust_graphql::handler::GraphQLState;
tokio::spawn(async move {
while reloads.changed().await.is_ok() {
let cache = Arc::new(state.schema_cache().await.clone());
config.enum_values = read_enum_values(&state.pool, &cache, &config.names).await;
let mut next = match GraphQLState::new(state.pool.clone(), cache, config.clone()) {
Ok(next) => next,
Err(e) => {
error!(
"GraphQL schema rebuild failed, keeping the previous one: {}",
e
);
continue;
}
};
let previous = current.read().unwrap_or_else(|e| e.into_inner()).clone();
if let Err(e) = next.continue_subscriptions(previous.as_deref()).await {
tracing::warn!("Subscriptions on the rebuilt schema: {}", e);
}
*current.write().unwrap_or_else(|e| e.into_inner()) = Some(Arc::new(next));
info!("GraphQL schema rebuilt");
}
});
}
#[cfg(feature = "graphql")]
async fn read_enum_values(
pool: &sqlx::PgPool,
cache: &postrust_core::schema_cache::SchemaCache,
names: &postrust_graphql::names::NameOverrides,
) -> std::collections::HashMap<String, Vec<(String, Option<String>)>> {
let mut values_of = std::collections::HashMap::new();
for (schema, table) in names.enum_tables() {
let qi = postrust_core::api_request::QualifiedIdentifier::new(&schema, &table);
let Some(definition) = cache.get_table(&qi) else {
tracing::warn!(
"{}.{} is marked as an enumeration but was not found",
schema,
table
);
continue;
};
let [key_column] = definition.pk_cols.as_slice() else {
tracing::warn!(
"{}.{} is marked as an enumeration but its primary key is not one column",
schema,
table
);
continue;
};
match postrust_core::schema_cache::enum_values(
pool,
&schema,
&table,
key_column,
definition.get_column("comment").is_some(),
)
.await
{
Ok(values) => {
values_of.insert(format!("{}.{}", schema, table), values);
}
Err(e) => {
tracing::warn!("cannot read the values of {}.{}: {}", schema, table, e)
}
}
}
values_of
}
fn mask_db_uri(uri: &str) -> String {
if let Some(at_pos) = uri.find('@') {
if let Some(proto_end) = uri.find("://") {
return format!("{}://***@{}", &uri[..proto_end], &uri[at_pos + 1..]);
}
}
uri.to_string()
}
#[cfg(test)]
mod tests {
use super::*;
fn routes(prefixes: &[&str]) -> Vec<(String, Router<Arc<AppState>>)> {
prefixes
.iter()
.map(|prefix| (prefix.to_string(), Router::new()))
.collect()
}
fn refusal(prefixes: &[&str]) -> String {
match mount_points(routes(prefixes)) {
Ok(_) => panic!("{prefixes:?} was accepted"),
Err(e) => e.to_string(),
}
}
#[test]
fn a_prefix_is_normalized() {
for given in ["hooks", "/hooks", "/hooks/", " /hooks// "] {
let nests = mount_points(routes(&[given])).unwrap();
assert_eq!(nests[0].0, "/hooks", "{given:?}");
}
}
#[test]
fn the_root_and_malformed_prefixes_are_refused() {
for given in ["", "/", "//", "/a//b", "/{id}", "/files/{*rest}", "/*"] {
refusal(&[given]);
}
}
#[test]
fn a_prefix_overlapping_the_servers_own_paths_is_refused() {
for given in [
"/api",
"/api/hooks",
"/healthz",
"/v1",
"/v1/version",
"/v1/graphql/extra",
"/v1alpha1",
"/admin",
"/_",
"/_/health",
"/_/ready/deep",
] {
let refused = refusal(&[given]);
assert!(refused.contains("serves"), "{given:?}: {refused}");
}
}
#[test]
fn a_prefix_sharing_only_characters_with_a_reserved_one_is_accepted() {
let nests = mount_points(routes(&[
"/apis",
"/v1/other",
"/administration",
"/_/hooks",
"/__",
]))
.unwrap();
assert_eq!(nests.len(), 5);
}
#[test]
fn two_prefixes_that_overlap_are_refused() {
assert!(refusal(&["/hooks", "hooks/"]).contains("nested earlier"));
assert!(refusal(&["/hooks", "/hooks/stripe"]).contains("nested earlier"));
assert!(refusal(&["/hooks/stripe", "/hooks"]).contains("nested earlier"));
assert!(refusal(&["/_/hooks", "/_/hooks"]).contains("nested earlier"));
}
#[test]
fn what_is_accepted_does_not_panic_the_router() {
let stub = || Router::new().route("/", any(|| async {}));
let nests = mount_points(
[
"/_/hooks",
"/_/jobs/nightly",
"/hooks",
"/v1/other",
"/apis",
]
.into_iter()
.map(|prefix| (prefix.to_string(), stub()))
.collect(),
)
.unwrap();
let mut app: Router<Arc<AppState>> = Router::new()
.nest("/api", Router::new().route("/{*path}", any(|| async {})))
.route("/healthz", any(|| async {}))
.route("/v1/version", any(|| async {}))
.nest(CUSTOM_PREFIX, custom::custom_router());
for (prefix, router) in nests {
app = app.nest(&prefix, router);
}
let _ = app
.nest("/admin", stub())
.nest("/v1/graphql", stub())
.nest("/v1alpha1/graphql", stub())
.nest("/api/graphql", stub())
.route("/", any(|| async {}))
.fallback(|| async {});
}
}