Skip to main content

postrust_server/
build.rs

1//! Building the application a host serves.
2//!
3//! Everything between a loaded configuration and a router answering requests:
4//! the pool, the schema cache, the REST surface, GraphQL and the layers around
5//! them. The `postrust` binary serves the result on a socket; the Lambda
6//! adapter hands the same router to the Lambda runtime. One path, so a request
7//! is answered the same way wherever it arrives -- the Lambda adapter used to
8//! be a second copy of the request path, and that copy bound no parameters and
9//! switched to no role.
10
11use crate::app::{self, handle_request};
12use crate::custom;
13use crate::state::AppState;
14use anyhow::Result;
15#[cfg(feature = "graphql")]
16use axum::routing::{get, post};
17use axum::{
18    extract::{DefaultBodyLimit, Request},
19    http::{HeaderValue, Method},
20    response::{IntoResponse, Json},
21    routing::{any, Route},
22    Router,
23};
24use sqlx::postgres::{PgConnectOptions, PgPoolOptions};
25use std::convert::Infallible;
26use std::sync::Arc;
27use tower::{Layer, Service};
28use tower_http::cors::{AllowOrigin, Any as CorsAny, CorsLayer};
29use tracing::{error, info};
30
31/// What the host running the application can do beyond answering requests.
32///
33/// A server process lives as long as it likes and owns its sockets. A Lambda
34/// sandbox is frozen between invocations, so a task listening for `NOTIFY` is
35/// asleep exactly when the notification arrives, and API Gateway does not
36/// upgrade a connection to a WebSocket at all.
37#[derive(Clone, Copy, Debug)]
38#[non_exhaustive]
39pub struct Host {
40    /// Long-lived background work: the `NOTIFY` schema reloader
41    /// (`db_channel_enabled`) and the subscription broker's listener.
42    pub background_tasks: bool,
43    /// Upgrading a connection to a WebSocket, which GraphQL subscriptions are
44    /// served over.
45    pub websockets: bool,
46}
47
48impl Host {
49    /// A long-running server process.
50    pub const SERVER: Host = Host {
51        background_tasks: true,
52        websockets: true,
53    };
54
55    /// An AWS Lambda function: no background work, no WebSockets.
56    pub const LAMBDA: Host = Host {
57        background_tasks: false,
58        websockets: false,
59    };
60}
61
62/// The application, ready to serve.
63#[non_exhaustive]
64pub struct App {
65    /// Every route the configuration enables, with its layers and state.
66    pub router: Router,
67    /// The state the routes share, for anything served beside them -- the
68    /// binary's admin port answers health checks from it.
69    pub state: Arc<AppState>,
70}
71
72/// Connect, load the schema cache and build the router.
73///
74/// Fails rather than serving something partial when the database cannot be
75/// reached or a setting cannot be honoured. GraphQL is the exception: a schema
76/// that cannot be built is logged and the rest is served without it.
77///
78/// The application with nothing of the caller's added. [`Builder`] builds the
79/// same with routes of the caller's own beside it.
80pub async fn build(config: postrust_core::AppConfig, host: Host) -> Result<App> {
81    Builder::new(config, host).build().await
82}
83
84/// Builds the application with routes of the caller's own beside the API.
85///
86/// For a binary of your own that depends on this crate rather than a fork of
87/// it: webhooks, an endpoint that does several things in one transaction,
88/// anything that is not a table or a function.
89///
90/// ```no_run
91/// # async fn run() -> anyhow::Result<()> {
92/// use axum::{routing::post, Router};
93/// use postrust_server::{AppState, Builder, Host};
94/// use std::sync::Arc;
95///
96/// let hooks: Router<Arc<AppState>> = Router::new().route("/stripe", post(|| async { "ok" }));
97/// let config = postrust_core::AppConfig::try_from_env()?;
98/// let app = Builder::new(config, Host::SERVER)
99///     .nest("/hooks", hooks)
100///     .build()
101///     .await?;
102/// # let _ = app;
103/// # Ok(())
104/// # }
105/// ```
106///
107/// A nested router is served with everything the API is served with: the
108/// CORS policy, the body limit, the `OPTIONS` handling and the shared
109/// [`AppState`]. Its routes are explicit routes, so in compatibility mode they
110/// are matched before the REST surface served at the root -- a table that
111/// shares a prefix's name is shadowed there, and stays reachable under `/api`.
112///
113/// [`Builder::layer`] is for what has to see every request rather than only
114/// the caller's own: an authentication gate, a rate limit, a header every
115/// response carries.
116#[must_use = "a Builder does nothing until `build` is called"]
117pub struct Builder {
118    config: postrust_core::AppConfig,
119    host: Host,
120    nests: Vec<(String, Router<Arc<AppState>>)>,
121    layers: Vec<Wrap>,
122}
123
124/// A layer given to [`Builder::layer`], held until there is a router for it.
125///
126/// Boxed as the step that applies it rather than as the layer, since each
127/// layer is its own type and only `Router::layer` knows what it needs of one.
128type Wrap = Box<dyn FnOnce(Router) -> Router + Send>;
129
130impl Builder {
131    /// The application `config` asks for, on `host`, with nothing added yet.
132    pub fn new(config: postrust_core::AppConfig, host: Host) -> Self {
133        Builder {
134            config,
135            host,
136            nests: Vec::new(),
137            layers: Vec::new(),
138        }
139    }
140
141    /// Serve `router` under `prefix`. May be called more than once.
142    ///
143    /// The prefix is taken as a path: the leading slash is added if missing
144    /// and a trailing one dropped, so `"hooks"`, `"/hooks"` and `"/hooks/"`
145    /// are one prefix. Nothing is checked until [`Builder::build`], which
146    /// refuses a prefix that is the root, that overlaps a path the server
147    /// serves itself ([`RESERVED_PREFIXES`], and `/_` itself), or that
148    /// overlaps another prefix given here. In compatibility mode, where REST
149    /// is served at `/` as well, a prefix whose first segment is `rpc` or the
150    /// name of a table or view in an exposed schema takes those paths from
151    /// REST, and a warning says so at start-up. A prefix beneath `/_`, such as
152    /// `/_/hooks`, is served beside the built-in `/_/health` and `/_/ready`.
153    pub fn nest(mut self, prefix: impl Into<String>, router: Router<Arc<AppState>>) -> Self {
154        self.nests.push((prefix.into(), router));
155        self
156    }
157
158    /// Wrap the whole application in `layer`. May be called more than once.
159    ///
160    /// Outermost, around every route the server serves and around its own
161    /// layers too, so it sees each request first and each response last --
162    /// including a preflight the CORS policy answers itself, and a refusal
163    /// that never reaches a handler. Of several, the one given last is the
164    /// outermost, as with `Router::layer`.
165    ///
166    /// For something only the caller's own routes need, put the layer on the
167    /// router given to [`Builder::nest`] instead.
168    pub fn layer<L>(mut self, layer: L) -> Self
169    where
170        L: Layer<Route> + Clone + Send + Sync + 'static,
171        L::Service: Service<Request> + Clone + Send + Sync + 'static,
172        <L::Service as Service<Request>>::Response: IntoResponse + 'static,
173        <L::Service as Service<Request>>::Error: Into<Infallible> + 'static,
174        <L::Service as Service<Request>>::Future: Send + 'static,
175    {
176        self.layers
177            .push(Box::new(move |router: Router| router.layer(layer)));
178        self
179    }
180
181    /// Connect, load the schema cache and build the router.
182    ///
183    /// The prefixes given to [`Builder::nest`] are checked first, before
184    /// anything is connected to, so a clash is reported as an error rather
185    /// than as a panic from the router -- and without a database to find it.
186    /// Otherwise as [`build`].
187    pub async fn build(self) -> Result<App> {
188        let nests = mount_points(self.nests)?;
189        // The JWT keys too, for the same reason: a key set that does not
190        // parse, or a JWKS URL that would not be fetched, is a configuration
191        // error, and is better reported before the database is connected to
192        // than after. Made once, here, and handed to the state.
193        let jwt = jwt_config(&self.config)?;
194        let mut app = assemble(self.config, self.host, nests, jwt).await?;
195        for wrap in self.layers {
196            app.router = wrap(app.router);
197        }
198        Ok(app)
199    }
200}
201
202/// The paths the server serves itself, which a prefix given to
203/// [`Builder::nest`] may neither equal nor sit above or beneath.
204///
205/// Reserved whether or not the feature that serves them is compiled in, so
206/// that a binary which builds today does not start failing when GraphQL or the
207/// admin UI is turned on. `/_` is not in the list: the built-in routes under
208/// it are, and the rest of it is the caller's.
209pub const RESERVED_PREFIXES: &[&str] = &[
210    "/api",
211    "/healthz",
212    "/v1/version",
213    "/v1/graphql",
214    "/v1alpha1/graphql",
215    "/admin",
216    "/_/health",
217    "/_/ready",
218];
219
220/// Where the built-in custom router is mounted.
221const CUSTOM_PREFIX: &str = "/_";
222
223/// The caller's routers, each under its prefix once the prefix is checked.
224type Nests = Vec<(String, Router<Arc<AppState>>)>;
225
226/// Check the prefixes given to [`Builder::nest`], normalizing each.
227fn mount_points(nests: Nests) -> Result<Nests> {
228    let mut checked: Nests = Vec::with_capacity(nests.len());
229    for (given, router) in nests {
230        let taken: Vec<&str> = checked.iter().map(|(prefix, _)| prefix.as_str()).collect();
231        let prefix = check_prefix(&given, &taken)?;
232        checked.push((prefix, router));
233    }
234    Ok(checked)
235}
236
237/// `given` as a prefix to nest at -- one leading slash, no trailing one -- or
238/// why it cannot be nested at beside the server's own paths and the prefixes
239/// already `taken`.
240fn check_prefix(given: &str, taken: &[&str]) -> Result<String> {
241    let trimmed = given.trim().trim_matches('/');
242    if trimmed.is_empty() {
243        // The root is the API's in compatibility mode and the server's own
244        // directory otherwise; a router merged there would take one or the
245        // other.
246        anyhow::bail!("cannot nest routes at {given:?}: a prefix must name a path below /");
247    }
248    // An empty segment matches nothing a client sends, and a parameter or a
249    // wildcard in a nest path is something the router refuses by panicking.
250    // Both are refused here instead, where it can be said why.
251    if trimmed
252        .split('/')
253        .any(|segment| segment.is_empty() || segment.contains(['{', '}', '*']))
254    {
255        anyhow::bail!(
256            "cannot nest routes at {given:?}: a prefix is literal path segments, \
257             without parameters, wildcards or empty segments"
258        );
259    }
260    let prefix = format!("/{trimmed}");
261
262    // Overlap rather than equality. `/v1` is not `/v1/version`, but nesting
263    // at `/v1` claims everything beneath it, and which of the two then
264    // answers `/v1/version` would be settled by the router's matching
265    // priority rather than by anything the caller wrote. A nest at exactly
266    // the same path as another is one the router refuses by panicking.
267    if prefix == CUSTOM_PREFIX {
268        anyhow::bail!(
269            "cannot nest routes at {given:?}: the server serves /_ itself; nest beneath it, \
270             at /_/<name>, instead"
271        );
272    }
273    if let Some(reserved) = RESERVED_PREFIXES
274        .iter()
275        .find(|reserved| overlaps(&prefix, reserved))
276    {
277        anyhow::bail!(
278            "cannot nest routes at {given:?}: it overlaps {reserved}, which the server serves itself"
279        );
280    }
281    if let Some(other) = taken.iter().find(|other| overlaps(&prefix, other)) {
282        anyhow::bail!("cannot nest routes at {given:?}: it overlaps {other}, nested earlier");
283    }
284    Ok(prefix)
285}
286
287/// Whether `path` is `base` or beneath it, segment by segment: `/apis` is not
288/// beneath `/api`.
289/// In compatibility mode REST is served at `/` too, beneath every path no
290/// route claims, so a prefix naming a table, a view or `/rpc` takes those
291/// requests from REST. That is allowed -- the route was asked for, and the
292/// table is still served under `/api` -- but said, so it is not found out
293/// from a response. Checked against the schema cache at start-up.
294fn warn_of_shadowing(nests: &Nests, schema_cache: &postrust_core::SchemaCache, schemas: &[String]) {
295    for (prefix, _) in nests {
296        let first = prefix
297            .trim_start_matches('/')
298            .split('/')
299            .next()
300            .unwrap_or_default();
301        if first == "rpc" {
302            tracing::warn!(
303                "compatibility mode: routes nested at {prefix} answer in place of the \
304                 database functions REST serves at /rpc; they are still served at /api/rpc"
305            );
306        }
307        if let Some(table) = schema_cache
308            .tables
309            .keys()
310            .find(|qi| qi.name == first && schemas.contains(&qi.schema))
311        {
312            tracing::warn!(
313                "compatibility mode: routes nested at {prefix} answer in place of {}.{first}, \
314                 which REST would serve at /{first}; it is still served at /api/{first}",
315                table.schema
316            );
317        }
318    }
319}
320
321fn beneath(path: &str, base: &str) -> bool {
322    path.strip_prefix(base)
323        .is_some_and(|rest| rest.is_empty() || rest.starts_with('/'))
324}
325
326/// Whether nesting at both `a` and `b` would claim a path twice.
327fn overlaps(a: &str, b: &str) -> bool {
328    beneath(a, b) || beneath(b, a)
329}
330
331/// The keys and claims tokens are verified with: `PGRST_JWT_SECRET`, and the
332/// key set at `PGRST_JWT_JWKS_URL` when there is one (not yet fetched).
333fn jwt_config(config: &postrust_core::AppConfig) -> Result<postrust_auth::JwtConfig> {
334    let jwt = postrust_engine::jwt_config(config)
335        .map_err(|e| anyhow::anyhow!("PGRST_JWT_SECRET: {e}"))?;
336    // Its errors name `PGRST_JWT_JWKS_URL` themselves.
337    crate::jwks::attach(jwt, config).map_err(|e| anyhow::anyhow!("{e}"))
338}
339
340/// Everything [`Builder::build`] does once the caller's prefixes and the JWT
341/// configuration are checked.
342async fn assemble(
343    config: postrust_core::AppConfig,
344    host: Host,
345    nests: Nests,
346    jwt: postrust_auth::JwtConfig,
347) -> Result<App> {
348    info!("Database: {}", mask_db_uri(&config.db_uri));
349
350    // Create database pool
351    let connect_options: PgConnectOptions = config
352        .db_uri
353        .parse()
354        .map_err(|e| anyhow::anyhow!("invalid database URI: {e}"))?;
355
356    // Prepared statements are sqlx's default and stay on unless asked. Turning
357    // them off is for a connection pooler in transaction mode -- PgBouncer and
358    // the like -- where the server connection a statement was prepared on is
359    // not the one the next query lands on, and every reuse fails with
360    // "prepared statement does not exist".
361    let connect_options = if config.db_prepared_statements {
362        connect_options
363    } else {
364        info!("Prepared statements disabled");
365        connect_options.statement_cache_capacity(0)
366    };
367
368    let pool = PgPoolOptions::new()
369        .max_connections(config.db_pool_size)
370        .acquire_timeout(std::time::Duration::from_secs(config.db_pool_timeout))
371        .connect_with(connect_options)
372        .await?;
373
374    info!("Connected to database");
375
376    // Load schema cache
377    let schema_cache = postrust_core::SchemaCache::load_with_search_path(
378        &pool,
379        &config.db_schemas,
380        &config.db_extra_search_path,
381    )
382    .await?;
383    info!("{}", schema_cache.summary());
384    if config.compat_mode {
385        warn_of_shadowing(&nests, &schema_cache, &config.db_schemas);
386    }
387
388    // Create app state
389    let state = Arc::new(AppState::with_jwt(pool, schema_cache, config.clone(), jwt));
390
391    // The pool and the token cache, for the metrics read when collected.
392    #[cfg(feature = "otel")]
393    crate::otel::observe(&state);
394
395    // `db_channel_enabled`: reload the schema cache when PostgreSQL says the
396    // schema changed, instead of on a restart. A migration ends with
397    // `NOTIFY pgrst, 'reload schema'` and the API picks the new shape up.
398    //
399    // The REST cache is swapped here; each reload then bumps `reloaded`, and
400    // the GraphQL schema, built from its own snapshot of the cache, is
401    // rebuilt from the new one when it sees the bump.
402    // Keys from `PGRST_JWT_JWKS_URL`: fetched once before serving, then kept
403    // fresh in the background where the host runs background work. A fetch
404    // that fails does not stop the start; see `crate::jwks`.
405    crate::jwks::start(&state, host).await;
406
407    let (reloaded, schema_reloads) = tokio::sync::watch::channel(0u64);
408    if config.db_channel_enabled && host.background_tasks {
409        spawn_schema_reloader(state.clone(), config.db_channel.clone(), reloaded);
410    }
411    #[cfg(not(feature = "graphql"))]
412    drop(schema_reloads);
413
414    // Build REST API router (under /api prefix)
415    let api_router: Router<Arc<AppState>> = Router::new()
416        .route("/", any(handle_request))
417        .route("/{*path}", any(handle_request));
418
419    // Build main router
420    let mut app: Router<Arc<AppState>> = Router::new().nest("/api", api_router);
421
422    // Two endpoints every Hasura deployment has, and that the things around a
423    // deployment reach for without being asked to: `/healthz` is what a load
424    // balancer, a Kubernetes probe and `docker-compose` healthchecks are
425    // already configured to poll, and `/v1/version` is what a client library
426    // calls to decide which features to use. Serving them costs nothing and
427    // not serving them makes a drop-in replacement fail its first health
428    // check.
429    app = app
430        .route("/healthz", axum::routing::get(|| async { "OK" }))
431        .route(
432            "/v1/version",
433            axum::routing::get(|| async {
434                Json(serde_json::json!({ "version": env!("CARGO_PKG_VERSION") }))
435            }),
436        );
437
438    // Add custom routes (health checks, webhooks, etc.)
439    //
440    // The caller's own routes go in here too, before any layer is applied, so
441    // they are wrapped in the same CORS policy, body limit and `OPTIONS`
442    // handling as the API. One beneath `/_`, such as `/_/hooks`, sits beside
443    // the nest at `/_`: the more specific path is matched first.
444    app = app.nest(CUSTOM_PREFIX, custom::custom_router());
445    info!("Custom routes enabled at /_");
446    for (prefix, router) in nests {
447        info!("Custom routes enabled at {}", prefix);
448        // In compatibility mode a database error is answered in the
449        // database's own words, by REST and so by a route's
450        // `postrust_server::Error` too. The error cannot see the
451        // configuration when `?` makes it, so this renders it again on the
452        // way out; see `crate::error`.
453        let router = match config.compat_mode {
454            true => router.layer(axum::middleware::map_response_with_state(
455                state.clone(),
456                crate::error::render_verbatim,
457            )),
458            false => router,
459        };
460        app = app.nest(&prefix, router);
461    }
462
463    // Add admin routes and GraphQL endpoint if feature is enabled
464    #[cfg(feature = "graphql")]
465    {
466        use async_graphql_axum::GraphQLBatchRequest as GqlRequest;
467        use axum::extract::State as AxumState;
468        use axum::http::HeaderMap;
469        use postrust_graphql::handler::GraphQLState;
470        use postrust_graphql::schema::SchemaConfig;
471
472        #[cfg(feature = "admin-ui")]
473        {
474            info!("Admin UI enabled at /admin");
475            app = app.nest("/admin", crate::admin::admin_router());
476        }
477
478        // Create GraphQL state with subscriptions enabled
479        let schema_cache_snapshot = state.schema_cache.read().await.clone();
480        let schema_cache_arc = Arc::new(schema_cache_snapshot);
481        // What Hasura keeps in metadata and a schema cannot carry: the names,
482        // and what each role may do. Absent, every name is derived exactly as
483        // before and there is no permission layer at all.
484        //
485        // Two spellings because the document outgrew the first one. `_NAMES`
486        // is what it was called when names were all it held, and a deployment
487        // already setting it should not have to change anything.
488        let names_var = ["PGRST_GRAPHQL_METADATA", "PGRST_GRAPHQL_NAMES"]
489            .into_iter()
490            .find_map(|name| std::env::var(name).ok().map(|value| (name, value)));
491
492        let graphql_names = match names_var {
493            Some((var, value)) => match postrust_graphql::names::NameOverrides::parse(&value) {
494                Ok(names) => {
495                    if !names.is_empty() {
496                        info!("GraphQL names given for {} tables", names.len());
497                    }
498                    if names.placed_functions() > 0 {
499                        info!(
500                            "GraphQL roots given for {} functions",
501                            names.placed_functions()
502                        );
503                    }
504                    if names.has_permissions() {
505                        info!(
506                            "GraphQL permissions given for {} roles, {} grants",
507                            names.roles().len(),
508                            names.granted()
509                        );
510                    }
511                    names
512                }
513                Err(e) => {
514                    // Serving the derived names instead would answer every
515                    // request under a name the client does not send, which
516                    // reads as a broken server rather than a bad setting. A
517                    // permission read wrong is worse still: it would serve
518                    // rows a rule was written to withhold.
519                    tracing::error!("{}: {}", var, e);
520                    return Err(anyhow::anyhow!("{}: {}", var, e));
521                }
522            },
523            None => postrust_graphql::names::NameOverrides::default(),
524        };
525
526        // An enum table's members are rows, not schema, so they are read here
527        // rather than reflected, and again each time the schema is rebuilt:
528        // a migration that adds a value notifies like one that adds a column.
529        let graphql_enum_values = read_enum_values(
530            &state.pool,
531            &*state.schema_cache.read().await,
532            &graphql_names,
533        )
534        .await;
535
536        // How often a live query re-reads itself with nothing having
537        // notified it. The notifications do the work; this is the floor
538        // under what a trigger cannot see -- a view, an embedded row on a
539        // table with no trigger, a predicate written against the clock.
540        let subscription_refresh_seconds = std::env::var("PGRST_SUBSCRIPTION_REFRESH")
541            .ok()
542            .and_then(|value| value.trim().parse::<u64>().ok())
543            .unwrap_or(30);
544
545        let mut graphql_config = SchemaConfig::default();
546        // A subscription is a WebSocket, so a host that cannot hold one
547        // does not advertise a `Subscription` root it cannot serve.
548        graphql_config.enable_subscriptions = host.websockets;
549        graphql_config.subscription_refresh_seconds = subscription_refresh_seconds;
550        graphql_config.max_rows = config.db_max_rows;
551        graphql_config.enable_federation = config.graphql_federation;
552        graphql_config.type_prefix = config.graphql_type_prefix.clone();
553        // The GraphQL schema was built for `public` whatever the server
554        // was told to expose, so a table in any other schema of
555        // `PGRST_DB_SCHEMAS` was reachable over REST and invisible over
556        // GraphQL.
557        graphql_config.exposed_schemas = config.db_schemas.clone();
558        graphql_config.enum_values = graphql_enum_values;
559        graphql_config.names = graphql_names;
560        // A schema that cannot be built is a reason to serve without GraphQL,
561        // not a reason to exit. The REST surface, the admin UI and the health
562        // endpoint do not depend on it, and taking the process down with it
563        // turns one unrepresentable table into a server that will not start.
564        // The endpoint is mounted all the same, answering that there is no
565        // schema, so the reload after a migration that fixes the table brings
566        // GraphQL up without a restart.
567        let graphql_state = match GraphQLState::new(
568            state.pool.clone(),
569            schema_cache_arc.clone(),
570            graphql_config.clone(),
571        ) {
572            Ok(graphql_state) => {
573                // Initialize subscription broker
574                if !host.websockets {
575                    info!("GraphQL subscriptions disabled: this host cannot hold a WebSocket");
576                } else if let Err(e) = graphql_state.init_subscriptions().await {
577                    tracing::warn!("Failed to initialize subscription broker: {}. Subscriptions may not work until triggers are created.", e);
578                } else {
579                    info!("GraphQL subscriptions enabled");
580                }
581                info!("GraphQL endpoint enabled at /api/graphql");
582                Some(Arc::new(graphql_state))
583            }
584            Err(e) => {
585                tracing::error!(
586                    "GraphQL schema could not be built, serving without it: {}",
587                    e
588                );
589                None
590            }
591        };
592        {
593            // The schema in use, replaced whole when the database's changes.
594            // A request takes the current one once and keeps it, so an
595            // operation is never answered half from one schema and half from
596            // the next.
597            let current = Arc::new(std::sync::RwLock::new(graphql_state));
598            spawn_graphql_rebuilder(
599                state.clone(),
600                current.clone(),
601                graphql_config,
602                schema_reloads,
603            );
604
605            // Combined state for GraphQL routes (includes JWT config for auth)
606            #[derive(Clone)]
607            struct GraphQLAppState {
608                gql_state: Arc<std::sync::RwLock<Option<Arc<GraphQLState>>>>,
609                /// REST's state, for the JWT configuration and the cache of
610                /// verified tokens: one token is verified once, whichever
611                /// surface it is presented to.
612                rest: Arc<AppState>,
613                hasura_auth: postrust_auth::HasuraAuthConfig,
614                /// REST's configuration, shared by every operation's context:
615                /// a GraphQL transaction is set up from it as a REST one is.
616                /// One copy for the server's life, since the configuration is
617                /// read once at start-up and never changes under it.
618                config: Arc<postrust_core::AppConfig>,
619            }
620
621            impl GraphQLAppState {
622                /// The GraphQL state to answer this operation from, if there
623                /// is one yet.
624                fn gql(&self) -> Option<Arc<GraphQLState>> {
625                    self.gql_state
626                        .read()
627                        .unwrap_or_else(|e| e.into_inner())
628                        .clone()
629                }
630            }
631
632            let graphql_app_state = GraphQLAppState {
633                gql_state: current,
634                rest: state.clone(),
635                hasura_auth: postrust_auth::HasuraAuthConfig {
636                    admin_secret: config.hasura_admin_secret.clone(),
637                    unauthorized_role: config.hasura_unauthorized_role.clone(),
638                },
639                config: Arc::new(state.config.clone()),
640            };
641
642            // Wrapper handler that creates context from request with proper auth
643            /// A body that is a JSON *array* is a batch: several operations
644            /// in one request, answered with an array of responses in the
645            /// order they were sent. Each is its own operation with its own
646            /// transaction -- batching is a way to save round trips, not a
647            /// way to make several mutations atomic, which is what naming
648            /// several root fields in one mutation is for.
649            async fn handle_graphql(
650                AxumState(app_state): AxumState<GraphQLAppState>,
651                // Which of the three addresses this arrived at. The routes are
652                // nested, so the request's own path is `/` by the time it gets
653                // here and the original is the only thing that still says.
654                axum::extract::OriginalUri(uri): axum::extract::OriginalUri,
655                headers: HeaderMap,
656                req: GqlRequest,
657            ) -> (axum::http::StatusCode, Json<serde_json::Value>) {
658                // The status a refused write is answered with belongs to the
659                // endpoint rather than to the refusal: `/v1alpha1/graphql`
660                // reports it in the transport and `/v1/graphql` does not.
661                let endpoint = match uri.path().starts_with("/v1alpha1/") {
662                    true => postrust_graphql::hasura::Endpoint::Legacy,
663                    false => postrust_graphql::hasura::Endpoint::Current,
664                };
665                match req.0 {
666                    async_graphql::BatchRequest::Single(request) => {
667                        let body = one_graphql(app_state, headers, request).await;
668                        let status = axum::http::StatusCode::from_u16(
669                            postrust_graphql::hasura::status_for(&body, endpoint),
670                        )
671                        .unwrap_or(axum::http::StatusCode::OK);
672                        (status, Json(body))
673                    }
674                    async_graphql::BatchRequest::Batch(requests) => {
675                        let mut answers = Vec::with_capacity(requests.len());
676                        for request in requests {
677                            answers.push(
678                                one_graphql(app_state.clone(), headers.clone(), request).await,
679                            );
680                        }
681                        // A batch is answered 200 whatever its parts say: the
682                        // transport carried every one of them, and which of
683                        // them failed is in the array.
684                        (
685                            axum::http::StatusCode::OK,
686                            Json(serde_json::Value::Array(answers)),
687                        )
688                    }
689                }
690            }
691
692            /// Who is asking, settled once: per request over HTTP, and per
693            /// connection over a WebSocket, where the token arrives in the
694            /// `connection_init` payload rather than on every operation.
695            #[derive(Clone)]
696            struct GraphQLIdentity {
697                auth_result: postrust_auth::AuthResult,
698                hasura_role: Option<String>,
699                session: std::collections::HashMap<String, String>,
700                elevated: bool,
701                headers: HeaderMap,
702            }
703
704            /// An operation ready to run against the schema its role answers
705            /// from, and the transaction its writes will share.
706            struct PreparedOperation {
707                schema: async_graphql::dynamic::Schema,
708                request: async_graphql::Request,
709                write: postrust_graphql::context::SharedWrite,
710            }
711
712            /// What [`postrust_auth::ensure_keys`] is to REST: keys from a
713            /// JWKS URL fetched, if the token needs them, before the
714            /// synchronous [`graphql_identity`] verifies it.
715            async fn ensure_keys(app_state: &GraphQLAppState, headers: &HeaderMap) {
716                let auth_header = headers.get("authorization").and_then(|v| v.to_str().ok());
717                postrust_auth::ensure_keys(auth_header, &app_state.rest.jwt_config).await;
718            }
719
720            /// The caller's identity, or the Hasura-shaped body refusing it.
721            fn graphql_identity(
722                app_state: &GraphQLAppState,
723                headers: HeaderMap,
724            ) -> Result<GraphQLIdentity, serde_json::Value> {
725                // Extract auth header and authenticate
726                let auth_header = headers.get("authorization").and_then(|v| v.to_str().ok());
727
728                // Through the token cache where one is configured, as REST
729                // authenticates.
730                let jwt_config = &app_state.rest.jwt_config;
731                let token = match &app_state.rest.jwt_cache {
732                    Some(cache) => cache.authenticate(auth_header, jwt_config),
733                    None => postrust_auth::authenticate(auth_header, jwt_config),
734                };
735                // A token that was offered and verified. Not the same as
736                // `token.is_ok()`, which is also true of the anonymous
737                // fallback -- and an anonymous request is precisely the one
738                // that has not been authenticated.
739                let token_verified = auth_header.is_some() && token.is_ok();
740                // A token that was offered and is refused: REST answers each of
741                // these 401. Without a secret there is nothing to verify
742                // against, and the header is ignored as Hasura ignores it when
743                // JWT mode is off -- a router forwarding `Authorization` to
744                // every subgraph relies on that. With no header at all, and no
745                // anonymous role, the request goes on as it always has.
746                let token_refusal = match (&token, auth_header) {
747                    (Ok(_) | Err(postrust_auth::JwtError::SecretMissing), _) | (Err(_), None) => {
748                        None
749                    }
750                    // Verified, and naming no role, with no anonymous role to
751                    // fall back to. Served as anyone, it would be served as a
752                    // role nobody chose.
753                    (Err(postrust_auth::JwtError::NoIdentity), Some(_)) => Some(
754                        postrust_graphql::hasura::missing_role_claim(&jwt_config.role_claim_key),
755                    ),
756                    (Err(e), Some(_)) => {
757                        Some(postrust_graphql::hasura::invalid_jwt(&e.to_string()))
758                    }
759                };
760
761                let auth_result = match token {
762                    Ok(auth) => auth,
763                    Err(e) => {
764                        tracing::debug!("GraphQL auth failed: {}, using anon role", e);
765                        postrust_auth::AuthResult {
766                            role: jwt_config
767                                .anon_role
768                                .clone()
769                                .unwrap_or_else(|| "anon".to_string()),
770                            claims: std::collections::HashMap::new(),
771                        }
772                    }
773                };
774
775                tracing::debug!(
776                    "GraphQL request authenticated as role: {}",
777                    auth_result.role
778                );
779
780                // Who the caller is in Hasura's sense, which is a different
781                // question from which database user the transaction runs as.
782                // The secret goes first: a caller that tried to be an
783                // administrator and got the secret wrong is refused rather
784                // than asked for a token.
785                let pairs: Vec<(&str, &str)> = headers
786                    .iter()
787                    .filter_map(|(name, value)| {
788                        value.to_str().ok().map(|value| (name.as_str(), value))
789                    })
790                    .collect();
791
792                let (hasura_role, session, elevated) = match (
793                    postrust_auth::hasura::from_admin_secret(&app_state.hasura_auth, &pairs[..]),
794                    token_refusal,
795                ) {
796                    (postrust_auth::SecretOutcome::Accepted(identity), _) => {
797                        (Some(identity.role), identity.session, identity.elevated)
798                    }
799                    (postrust_auth::SecretOutcome::Rejected, _) => {
800                        return Err(postrust_graphql::hasura::denied(
801                            postrust_auth::hasura::SECRET_INVALID,
802                        ));
803                    }
804                    // The secret settled nothing, so the token is refused
805                    // -- and an administrator's secret beside it wins, as
806                    // it does in Hasura.
807                    (_, Some(refusal)) => return Err(refusal),
808                    // Nothing settled by the secret. A verified token speaks
809                    // next, and session variables come from its claims.
810                    //
811                    // One header is read here and only here: `X-Hasura-Role`,
812                    // against the `x-hasura-allowed-roles` the token carries.
813                    // That list is inside the signature, so it lets a caller
814                    // choose among the identities it was issued and not add
815                    // one. Every other `x-hasura-*` header is still ignored --
816                    // on a server with no secret to gate them they would let
817                    // any caller name its own identity.
818                    _ if token_verified => {
819                        match postrust_auth::hasura::role_for_token(&auth_result.claims, &pairs[..])
820                        {
821                            postrust_auth::hasura::TokenRole::Is(role) => (
822                                role,
823                                postrust_auth::hasura::session_from_claims(&auth_result.claims),
824                                false,
825                            ),
826                            postrust_auth::hasura::TokenRole::NotAllowed => {
827                                return Err(postrust_graphql::hasura::denied(
828                                    postrust_auth::hasura::ROLE_NOT_ALLOWED,
829                                ));
830                            }
831                        }
832                    }
833                    // A secret is configured and nothing authenticated this
834                    // request. Answering it as a stranger is a choice the
835                    // deployment makes; refusing is the default.
836                    _ if app_state.hasura_auth.is_configured() => {
837                        match postrust_auth::hasura::unauthenticated(&app_state.hasura_auth) {
838                            Some(identity) => (Some(identity.role), identity.session, false),
839                            None => {
840                                return Err(postrust_graphql::hasura::denied(
841                                    postrust_auth::hasura::SECRET_MISSING,
842                                ));
843                            }
844                        }
845                    }
846                    // No Hasura auth configured at all: unchanged from before
847                    // any of this existed.
848                    _ => (
849                        postrust_auth::hasura::role_from_claims(&auth_result.claims),
850                        postrust_auth::hasura::session_from_claims(&auth_result.claims),
851                        false,
852                    ),
853                };
854
855                Ok(GraphQLIdentity {
856                    auth_result,
857                    hasura_role,
858                    session,
859                    elevated,
860                    headers,
861                })
862            }
863
864            /// Everything between knowing who is asking and running what they
865            /// asked: the checks Hasura makes before reading the document, the
866            /// schema the role answers from, and the context every resolver
867            /// reads the identity from.
868            fn prepare_operation(
869                app_state: &GraphQLAppState,
870                identity: &GraphQLIdentity,
871                req: async_graphql::Request,
872            ) -> Result<PreparedOperation, serde_json::Value> {
873                let GraphQLIdentity {
874                    auth_result,
875                    hasura_role,
876                    session,
877                    elevated,
878                    headers,
879                } = identity.clone();
880                let Some(gql) = app_state.gql() else {
881                    return Err(postrust_graphql::hasura::unavailable(
882                        "the GraphQL schema could not be built; see the server log",
883                    ));
884                };
885                let pairs: Vec<(&str, &str)> = headers
886                    .iter()
887                    .filter_map(|(name, value)| {
888                        value.to_str().ok().map(|value| (name.as_str(), value))
889                    })
890                    .collect();
891
892                // A protocol neither server implements, which only one of
893                // them says. See `hasura::persisted_query`.
894                if postrust_graphql::hasura::persisted_query(&req) {
895                    return Err(postrust_graphql::hasura::not_supported(
896                        "PersistedQueryNotSupported",
897                    ));
898                }
899
900                // A header Hasura reads itself, before it reads the
901                // document. `x-hasura-use-backend-only-permissions: random` is
902                // not false, and treating it as false would send a
903                // backend-only write down the path meant for everyone else on
904                // a header the client thought it had set.
905                if let Some(message) = postrust_auth::hasura::unreadable_boolean_header(&pairs[..])
906                {
907                    return Err(postrust_graphql::hasura::malformed(&message));
908                }
909
910                // Which schema answers. A permission is a statement about what
911                // exists, so the role decides the shape of the API before it
912                // decides anything about a row: a role the document does not
913                // name has no schema at all, and is refused here rather than
914                // being answered from someone else's.
915                let Some(schema) = gql.schema_for(
916                    hasura_role.as_deref(),
917                    postrust_auth::hasura::backend_only_requested(&pairs[..], elevated),
918                ) else {
919                    return Err(postrust_graphql::hasura::denied(&format!(
920                        "role \"{}\" is not defined in the permissions",
921                        hasura_role.as_deref().unwrap_or_default()
922                    )));
923                };
924
925                // Create SchemaCacheRef from the static Arc<SchemaCache>
926                //
927                // The unreduced cache, deliberately. The schema above decides
928                // what may be asked for; this is what a resolver looks a type
929                // up in once something has been asked, and a superset can only
930                // answer the same questions.
931                let schema_cache_ref = postrust_core::schema_cache::SchemaCacheRef::from_static(
932                    (*gql.schema_cache).clone(),
933                );
934
935                // Every write in this operation goes into one transaction, and
936                // this is the half that decides its fate: a mutation naming
937                // several root fields is all-or-nothing, so the second one
938                // failing has to take the first one's rows with it. Nothing to
939                // settle for a query, which never opens it.
940                let write: postrust_graphql::context::SharedWrite = Arc::default();
941
942                let gql_ctx = postrust_graphql::context::GraphQLContext::new(
943                    gql.pool.clone(),
944                    schema_cache_ref,
945                    auth_result,
946                )
947                .with_session(session)
948                .with_identity(hasura_role.clone(), elevated)
949                .with_write(Arc::clone(&write))
950                .with_config(Arc::clone(&app_state.config));
951
952                // A role may be told it cannot read the schema as data. Not a
953                // permission on a table, so it is not settled by which schema
954                // answers -- the schema is the thing being withheld.
955                //
956                // Not withheld from a caller holding the admin secret, whatever
957                // role it then names. That is measured rather than assumed: a
958                // v2.50.1 reference given `set_graphql_schema_introspection_
959                // options` answers `__schema` for the named role when the
960                // secret is sent beside the role header, and answers it from
961                // that role's own restricted schema. Reading the schema is an
962                // administrator's to do; the permissions still apply to it.
963                let introspection_disabled_for = hasura_role
964                    .as_deref()
965                    .filter(|_| !elevated)
966                    .filter(|role| gql.config.names.introspection_disabled(role));
967                let prepared = postrust_graphql::hasura::prepare(
968                    Some(schema),
969                    req,
970                    introspection_disabled_for,
971                );
972                let request = match prepared {
973                    Ok(request) => request,
974                    // Refused before it ran: nothing was written, so there is
975                    // nothing to settle.
976                    Err((_, errors)) => {
977                        let mut response = async_graphql::Response::new(async_graphql::Value::Null);
978                        response.errors = errors;
979                        return Err(postrust_graphql::hasura::envelope(response));
980                    }
981                };
982                let request = request
983                    .data(gql_ctx)
984                    .data(gql.pool.clone())
985                    .data(Arc::clone(&gql.broker));
986                Ok(PreparedOperation {
987                    schema: schema.clone(),
988                    request,
989                    write,
990                })
991            }
992
993            /// Settle an operation's writes before its answer goes out.
994            async fn settle_write(
995                write: &postrust_graphql::context::SharedWrite,
996                response: &mut async_graphql::Response,
997            ) {
998                if let Some(tx) = write.lock().await.take() {
999                    let settled = if response.errors.is_empty() {
1000                        tx.commit().await
1001                    } else {
1002                        tx.rollback().await
1003                    };
1004                    // A commit that fails is a mutation that did not happen,
1005                    // however well every statement in it went. Saying so is
1006                    // the whole point of running them together.
1007                    if let Err(e) = settled {
1008                        tracing::error!("GraphQL mutation could not be settled: {}", e);
1009                        response
1010                            .errors
1011                            .push(async_graphql::ServerError::new(e.to_string(), None));
1012                        response.data = async_graphql::Value::Null;
1013                    }
1014                }
1015            }
1016
1017            async fn one_graphql(
1018                app_state: GraphQLAppState,
1019                headers: HeaderMap,
1020                req: async_graphql::Request,
1021            ) -> serde_json::Value {
1022                ensure_keys(&app_state, &headers).await;
1023                let identity = match graphql_identity(&app_state, headers) {
1024                    Ok(identity) => identity,
1025                    Err(refusal) => return refusal,
1026                };
1027                let PreparedOperation {
1028                    schema,
1029                    request,
1030                    write,
1031                } = match prepare_operation(&app_state, &identity, req) {
1032                    Ok(operation) => operation,
1033                    Err(refusal) => return refusal,
1034                };
1035                // One span for the operation, beneath the request's, named by
1036                // the operation as the GraphQL conventions have it. A span
1037                // and not async-graphql's OpenTelemetry extension: that is
1038                // built against opentelemetry 0.27, and this against 0.33.
1039                use tracing::Instrument as _;
1040                let span = tracing::debug_span!(
1041                    "graphql",
1042                    "graphql.operation.name" = request.operation_name.as_deref(),
1043                );
1044                let mut response = schema.execute(request).instrument(span).await;
1045                settle_write(&write, &mut response).await;
1046                postrust_graphql::hasura::envelope(response)
1047            }
1048
1049            /// A Hasura-shaped body as the response a WebSocket carries, so a
1050            /// subscription is refused, and its errors read, exactly as the
1051            /// same operation over HTTP would be.
1052            fn hasura_response(body: serde_json::Value) -> async_graphql::Response {
1053                let data = body
1054                    .get("data")
1055                    .cloned()
1056                    .and_then(|data| async_graphql::Value::from_json(data).ok())
1057                    .unwrap_or(async_graphql::Value::Null);
1058                let mut response = async_graphql::Response::new(data);
1059                for error in body
1060                    .get("errors")
1061                    .and_then(serde_json::Value::as_array)
1062                    .into_iter()
1063                    .flatten()
1064                {
1065                    let message = error
1066                        .get("message")
1067                        .and_then(serde_json::Value::as_str)
1068                        .unwrap_or_default();
1069                    let mut server_error = async_graphql::ServerError::new(message, None);
1070                    if let Some(fields) = error
1071                        .get("extensions")
1072                        .and_then(serde_json::Value::as_object)
1073                    {
1074                        let mut extensions = async_graphql::ErrorExtensionValues::default();
1075                        for (name, value) in fields {
1076                            if let Ok(value) = async_graphql::Value::from_json(value.clone()) {
1077                                extensions.set(name, value);
1078                            }
1079                        }
1080                        server_error.extensions = Some(extensions);
1081                    }
1082                    response.errors.push(server_error);
1083                }
1084                response
1085            }
1086
1087            /// The headers a WebSocket connection is authenticated with: the
1088            /// upgrade request's, overlaid with those the `connection_init`
1089            /// payload carries. A browser cannot set headers on a WebSocket,
1090            /// so a client puts `Authorization` in the payload -- Hasura's own
1091            /// clients under `headers`, others at the top level.
1092            fn connection_headers(
1093                mut headers: HeaderMap,
1094                payload: &serde_json::Value,
1095            ) -> HeaderMap {
1096                let nested = payload
1097                    .get("headers")
1098                    .and_then(serde_json::Value::as_object);
1099                for fields in [payload.as_object(), nested].into_iter().flatten() {
1100                    for (name, value) in fields {
1101                        let Some(value) = value.as_str() else {
1102                            continue;
1103                        };
1104                        if let (Ok(name), Ok(value)) = (
1105                            axum::http::HeaderName::from_bytes(name.as_bytes()),
1106                            axum::http::HeaderValue::from_str(value),
1107                        ) {
1108                            headers.insert(name, value);
1109                        }
1110                    }
1111                }
1112                headers
1113            }
1114
1115            /// Runs every WebSocket operation -- query, mutation or
1116            /// subscription -- as the identity its connection was initialised
1117            /// with, through the same preparation an HTTP request gets.
1118            #[derive(Clone)]
1119            struct WebSocketExecutor(GraphQLAppState);
1120
1121            impl async_graphql::Executor for WebSocketExecutor {
1122                async fn execute(
1123                    &self,
1124                    request: async_graphql::Request,
1125                ) -> async_graphql::Response {
1126                    use async_graphql::futures_util::StreamExt;
1127                    // Not reached over a WebSocket, which runs everything as a
1128                    // stream; answered from the stream so the two cannot differ.
1129                    self.execute_stream(request, None)
1130                        .next()
1131                        .await
1132                        .unwrap_or_else(|| async_graphql::Response::new(async_graphql::Value::Null))
1133                }
1134
1135                fn execute_stream(
1136                    &self,
1137                    request: async_graphql::Request,
1138                    session_data: Option<Arc<async_graphql::Data>>,
1139                ) -> async_graphql::futures_util::stream::BoxStream<'static, async_graphql::Response>
1140                {
1141                    use async_graphql::futures_util::{stream, StreamExt};
1142                    let identity = session_data
1143                        .as_deref()
1144                        .and_then(|data| data.get(&std::any::TypeId::of::<GraphQLIdentity>()))
1145                        .and_then(|identity| identity.downcast_ref::<GraphQLIdentity>())
1146                        .cloned();
1147                    // `connection_init` refuses a connection it cannot
1148                    // identify, so an operation without an identity is one the
1149                    // protocol let through before initialising. Refused rather
1150                    // than run as nobody in particular.
1151                    let Some(identity) = identity else {
1152                        let refusal =
1153                            postrust_graphql::hasura::denied("the connection was not initialised");
1154                        return stream::once(async move { hasura_response(refusal) }).boxed();
1155                    };
1156                    match prepare_operation(&self.0, &identity, request) {
1157                        Err(refusal) => {
1158                            stream::once(async move { hasura_response(refusal) }).boxed()
1159                        }
1160                        Ok(PreparedOperation {
1161                            schema,
1162                            request,
1163                            write,
1164                        }) => {
1165                            async_graphql::Executor::execute_stream(&schema, request, session_data)
1166                                .then(move |mut response| {
1167                                    let write = Arc::clone(&write);
1168                                    async move {
1169                                        settle_write(&write, &mut response).await;
1170                                        hasura_response(postrust_graphql::hasura::envelope(
1171                                            response,
1172                                        ))
1173                                    }
1174                                })
1175                                .boxed()
1176                        }
1177                    }
1178                }
1179            }
1180
1181            /// GraphQL over a WebSocket, authenticated as HTTP is. The
1182            /// identity is settled once, from `connection_init`, and a
1183            /// connection that would have been refused over HTTP -- a wrong
1184            /// admin secret, a role the token does not allow -- is refused
1185            /// there rather than served as the anonymous role.
1186            async fn graphql_ws(
1187                AxumState(app_state): AxumState<GraphQLAppState>,
1188                protocol: async_graphql_axum::GraphQLProtocol,
1189                headers: HeaderMap,
1190                ws: axum::extract::ws::WebSocketUpgrade,
1191            ) -> axum::response::Response {
1192                ws.protocols(async_graphql::http::ALL_WEBSOCKET_PROTOCOLS)
1193                    .on_upgrade(move |socket| async move {
1194                        let init_state = app_state.clone();
1195                        async_graphql_axum::GraphQLWebSocket::new(
1196                            socket,
1197                            WebSocketExecutor(app_state),
1198                            protocol,
1199                        )
1200                        .on_connection_init(move |payload| async move {
1201                            let headers = connection_headers(headers, &payload);
1202                            ensure_keys(&init_state, &headers).await;
1203                            match graphql_identity(&init_state, headers) {
1204                                Ok(identity) => {
1205                                    let mut data = async_graphql::Data::default();
1206                                    data.insert(identity);
1207                                    Ok(data)
1208                                }
1209                                Err(refusal) => Err(async_graphql::Error::new(
1210                                    refusal
1211                                        .pointer("/errors/0/message")
1212                                        .and_then(serde_json::Value::as_str)
1213                                        .unwrap_or("connection refused"),
1214                                )),
1215                            }
1216                        })
1217                        .serve()
1218                        .await
1219                    })
1220            }
1221
1222            // Add GraphQL routes with WebSocket support for subscriptions
1223            let graphql_router = Router::new()
1224                .route("/", post(handle_graphql))
1225                .route("/", get(postrust_graphql::handler::graphql_playground))
1226                .with_state(graphql_app_state.clone());
1227
1228            // The same state as HTTP, because a WebSocket is authenticated the
1229            // same way: `postrust_graphql::handler::graphql_ws_handler` carries
1230            // no caller identity, so every subscription it served ran without
1231            // a role, a session or claims.
1232            let ws_router = Router::new()
1233                .route("/ws", get(graphql_ws))
1234                .with_state(graphql_app_state);
1235
1236            let graphql_app = match host.websockets {
1237                true => graphql_router.merge(ws_router),
1238                false => graphql_router,
1239            };
1240            // `/v1/graphql` is where a Hasura client sends its queries, and it is
1241            // the only address most of them can be told about: the endpoint is
1242            // baked into generated clients and codegen configs. `/api/graphql`
1243            // keeps working for anything already pointed at it, and
1244            // `/v1alpha1/graphql` is the address Hasura served before `/v1`
1245            // and still answers on -- a client old enough to have been
1246            // pointed there is exactly the one that cannot be repointed.
1247            app = app
1248                .nest("/v1/graphql", graphql_app.clone())
1249                .nest("/v1alpha1/graphql", graphql_app.clone())
1250                .nest("/api/graphql", graphql_app);
1251        }
1252    }
1253
1254    // PostgREST compatibility mode: also serve the REST surface at the root so
1255    // canonical PostgREST paths (`/rpc/<name>`, `/<table>`) work in addition to
1256    // the `/api`-prefixed paths. Explicit routes (`/`, `/_`, `/admin`, `/api`,
1257    // and whatever was given to `Builder::nest`) still take precedence; only
1258    // otherwise-unmatched paths hit this fallback.
1259    if config.compat_mode {
1260        app = app.fallback(handle_request);
1261        info!("PostgREST compatibility mode enabled: REST surface also served at /");
1262
1263        // Key ordering is fixed when the binary is compiled, so asking for
1264        // compatibility at runtime cannot turn it on. Say so, rather than
1265        // leaving someone to discover the difference by diffing responses.
1266        if !cfg!(feature = "compat-key-order") {
1267            tracing::warn!(
1268                "compatibility mode: object keys will be returned in alphabetical order, \
1269                 not in select order as PostgREST returns them. Build with \
1270                 --features compat-key-order to match, at a cost of up to 15% throughput \
1271                 on wide rows."
1272            );
1273        }
1274    }
1275
1276    // Add root info endpoint.
1277    //
1278    // Not in compatibility mode: there `/` is part of the API surface -- it is
1279    // where PostgREST serves the schema description, and where it reports an
1280    // `Accept-Profile` naming a schema that is not exposed. A directory of
1281    // this server's own endpoints in its place answers a different question
1282    // from the one asked.
1283    if !config.compat_mode {
1284        app = app.route(
1285            "/",
1286            axum::routing::get(|| async {
1287                Json(serde_json::json!({
1288                    "name": "postrust",
1289                    "version": env!("CARGO_PKG_VERSION"),
1290                    "api": "/api",
1291                    "custom": "/_",
1292                    "health": "/_/health",
1293                    "admin": "/admin",
1294                    "docs": "/admin/swagger"
1295                }))
1296            }),
1297        );
1298    }
1299
1300    // An empty list is "any origin", which is the default and what
1301    // `PGRST_SERVER_CORS_ORIGINS="*"` asks for. Anything that is not a valid
1302    // header value is dropped with a warning rather than quietly widening the
1303    // policy back to `*` -- a typo in an origin should cost that origin, not
1304    // the restriction.
1305    let allow_origin = if config.server_cors_origins.is_empty() {
1306        AllowOrigin::any()
1307    } else {
1308        let origins: Vec<HeaderValue> = config
1309            .server_cors_origins
1310            .iter()
1311            .filter_map(|o| match HeaderValue::from_str(o) {
1312                Ok(v) => Some(v),
1313                Err(_) => {
1314                    tracing::warn!("Ignoring CORS origin {:?}: not a valid header value", o);
1315                    None
1316                }
1317            })
1318            .collect();
1319        info!("CORS restricted to {} origin(s)", origins.len());
1320        AllowOrigin::list(origins)
1321    };
1322
1323    // Apply CORS and state
1324    let app = app
1325        .layer(
1326            CorsLayer::new()
1327                .allow_origin(allow_origin)
1328                .allow_methods([
1329                    Method::GET,
1330                    Method::POST,
1331                    Method::PUT,
1332                    Method::PATCH,
1333                    Method::DELETE,
1334                    Method::OPTIONS,
1335                    Method::HEAD,
1336                ])
1337                .allow_headers(CorsAny)
1338                .expose_headers(CorsAny),
1339        )
1340        // For the routes whose handlers take a body *extractor* -- GraphQL,
1341        // the admin surface -- which is what this layer reaches. The REST
1342        // handler takes the whole `Request` and reads the body itself, so it
1343        // applies `max_body_size` directly; see `postrust_engine::handle`.
1344        .layer(DefaultBodyLimit::max(config.max_body_size))
1345        // Outermost, so it runs before the CORS layer -- which answers every
1346        // OPTIONS itself and never calls what it wraps, so nothing downstream
1347        // of it can say what a resource allows.
1348        .layer(axum::middleware::from_fn_with_state(
1349            state.clone(),
1350            app::options_allow,
1351        ))
1352        // Outermost of all, so the request span covers everything above --
1353        // an OPTIONS answered by `options_allow`, a preflight answered by
1354        // CORS -- and every response, whoever made it, carries the request
1355        // id. A `Router::layer`, so the router has matched the route before
1356        // this runs and `MatchedPath` is there to name the span's
1357        // `http.route`. See `crate::telemetry`.
1358        .layer(crate::telemetry::RequestTraceLayer)
1359        .with_state(state.clone());
1360
1361    Ok(App { router: app, state })
1362}
1363
1364/// Reload the REST schema cache on `NOTIFY <channel>`.
1365///
1366/// Reconnects on its own: a listening connection is dropped by a restart of
1367/// the database or anything between, and a reloader that gives up on the first
1368/// error is one that stops working silently partway through a deployment.
1369fn spawn_schema_reloader(
1370    state: Arc<AppState>,
1371    channel: String,
1372    reloaded: tokio::sync::watch::Sender<u64>,
1373) {
1374    const RETRY: std::time::Duration = std::time::Duration::from_secs(5);
1375
1376    tokio::spawn(async move {
1377        loop {
1378            let mut listener = match sqlx::postgres::PgListener::connect_with(&state.pool).await {
1379                Ok(listener) => listener,
1380                Err(e) => {
1381                    error!("Schema reloader could not connect: {}; retrying", e);
1382                    tokio::time::sleep(RETRY).await;
1383                    continue;
1384                }
1385            };
1386
1387            if let Err(e) = listener.listen(&channel).await {
1388                error!("Schema reloader could not LISTEN on {}: {}", channel, e);
1389                tokio::time::sleep(RETRY).await;
1390                continue;
1391            }
1392            info!("Listening on channel {} for schema reloads", channel);
1393
1394            // Inner loop until the connection fails, then reconnect.
1395            loop {
1396                match listener.recv().await {
1397                    Ok(notification) => {
1398                        info!("Schema reload requested: {:?}", notification.payload());
1399                        match state.reload_schema().await {
1400                            Ok(()) => {
1401                                info!(
1402                                    "Schema cache reloaded: {}",
1403                                    state.schema_cache().await.summary()
1404                                );
1405                                reloaded.send_modify(|n| *n += 1);
1406                                #[cfg(feature = "otel")]
1407                                crate::otel::schema_reloaded(true);
1408                            }
1409                            // The old cache is kept. Serving the previous
1410                            // schema is better than serving none, and the next
1411                            // notification tries again.
1412                            Err(e) => {
1413                                error!("Schema cache reload failed: {}", e);
1414                                #[cfg(feature = "otel")]
1415                                crate::otel::schema_reloaded(false);
1416                            }
1417                        }
1418                    }
1419                    Err(e) => {
1420                        error!("Schema reloader lost its connection: {}; reconnecting", e);
1421                        break;
1422                    }
1423                }
1424            }
1425        }
1426    });
1427}
1428
1429/// Rebuild the GraphQL schema each time the REST schema cache is reloaded.
1430///
1431/// The new state is built beside the old one and swapped in only once it is
1432/// whole: a schema that cannot be built leaves the previous one serving, as a
1433/// failed reload of the REST cache does. Enum tables are read again, since a
1434/// migration may add a value; the permission document is kept from start-up,
1435/// since it is not part of what a migration notifies about. The subscription
1436/// broker carries over to the new state, so a subscription opened on the old
1437/// schema keeps receiving notifications, and listens on any table the new
1438/// one adds.
1439#[cfg(feature = "graphql")]
1440fn spawn_graphql_rebuilder(
1441    state: Arc<AppState>,
1442    current: Arc<std::sync::RwLock<Option<Arc<postrust_graphql::handler::GraphQLState>>>>,
1443    mut config: postrust_graphql::schema::SchemaConfig,
1444    mut reloads: tokio::sync::watch::Receiver<u64>,
1445) {
1446    use postrust_graphql::handler::GraphQLState;
1447
1448    tokio::spawn(async move {
1449        while reloads.changed().await.is_ok() {
1450            let cache = Arc::new(state.schema_cache().await.clone());
1451            config.enum_values = read_enum_values(&state.pool, &cache, &config.names).await;
1452            let mut next = match GraphQLState::new(state.pool.clone(), cache, config.clone()) {
1453                Ok(next) => next,
1454                Err(e) => {
1455                    error!(
1456                        "GraphQL schema rebuild failed, keeping the previous one: {}",
1457                        e
1458                    );
1459                    continue;
1460                }
1461            };
1462            let previous = current.read().unwrap_or_else(|e| e.into_inner()).clone();
1463            if let Err(e) = next.continue_subscriptions(previous.as_deref()).await {
1464                // The tables already listened to still are; a new one is
1465                // answered by the refresh until the next reload retries it.
1466                tracing::warn!("Subscriptions on the rebuilt schema: {}", e);
1467            }
1468            *current.write().unwrap_or_else(|e| e.into_inner()) = Some(Arc::new(next));
1469            info!("GraphQL schema rebuilt");
1470        }
1471    });
1472}
1473
1474/// The members of every table the GraphQL metadata marks as an enumeration,
1475/// keyed "schema.table". A table that cannot be read is left out, with a
1476/// warning, and becomes an ordinary table type.
1477#[cfg(feature = "graphql")]
1478async fn read_enum_values(
1479    pool: &sqlx::PgPool,
1480    cache: &postrust_core::schema_cache::SchemaCache,
1481    names: &postrust_graphql::names::NameOverrides,
1482) -> std::collections::HashMap<String, Vec<(String, Option<String>)>> {
1483    let mut values_of = std::collections::HashMap::new();
1484    for (schema, table) in names.enum_tables() {
1485        let qi = postrust_core::api_request::QualifiedIdentifier::new(&schema, &table);
1486        let Some(definition) = cache.get_table(&qi) else {
1487            tracing::warn!(
1488                "{}.{} is marked as an enumeration but was not found",
1489                schema,
1490                table
1491            );
1492            continue;
1493        };
1494        // One column identifies a member. A composite key names no
1495        // single value, so there is nothing to call the member.
1496        let [key_column] = definition.pk_cols.as_slice() else {
1497            tracing::warn!(
1498                "{}.{} is marked as an enumeration but its primary key is not one column",
1499                schema,
1500                table
1501            );
1502            continue;
1503        };
1504        // Hasura's convention, and a useful one: a `comment` column
1505        // describes each value. Read as a role that may read the
1506        // table, which a NOINHERIT authenticator is not by itself.
1507        match postrust_core::schema_cache::enum_values(
1508            pool,
1509            &schema,
1510            &table,
1511            key_column,
1512            definition.get_column("comment").is_some(),
1513        )
1514        .await
1515        {
1516            Ok(values) => {
1517                values_of.insert(format!("{}.{}", schema, table), values);
1518            }
1519            Err(e) => {
1520                tracing::warn!("cannot read the values of {}.{}: {}", schema, table, e)
1521            }
1522        }
1523    }
1524    values_of
1525}
1526
1527/// Mask database URI for logging.
1528fn mask_db_uri(uri: &str) -> String {
1529    if let Some(at_pos) = uri.find('@') {
1530        if let Some(proto_end) = uri.find("://") {
1531            return format!("{}://***@{}", &uri[..proto_end], &uri[at_pos + 1..]);
1532        }
1533    }
1534    uri.to_string()
1535}
1536
1537#[cfg(test)]
1538mod tests {
1539    use super::*;
1540
1541    fn routes(prefixes: &[&str]) -> Vec<(String, Router<Arc<AppState>>)> {
1542        prefixes
1543            .iter()
1544            .map(|prefix| (prefix.to_string(), Router::new()))
1545            .collect()
1546    }
1547
1548    fn refusal(prefixes: &[&str]) -> String {
1549        match mount_points(routes(prefixes)) {
1550            Ok(_) => panic!("{prefixes:?} was accepted"),
1551            Err(e) => e.to_string(),
1552        }
1553    }
1554
1555    #[test]
1556    fn a_prefix_is_normalized() {
1557        for given in ["hooks", "/hooks", "/hooks/", " /hooks// "] {
1558            let nests = mount_points(routes(&[given])).unwrap();
1559            assert_eq!(nests[0].0, "/hooks", "{given:?}");
1560        }
1561    }
1562
1563    #[test]
1564    fn the_root_and_malformed_prefixes_are_refused() {
1565        for given in ["", "/", "//", "/a//b", "/{id}", "/files/{*rest}", "/*"] {
1566            refusal(&[given]);
1567        }
1568    }
1569
1570    #[test]
1571    fn a_prefix_overlapping_the_servers_own_paths_is_refused() {
1572        for given in [
1573            "/api",
1574            "/api/hooks",
1575            "/healthz",
1576            "/v1",
1577            "/v1/version",
1578            "/v1/graphql/extra",
1579            "/v1alpha1",
1580            "/admin",
1581            "/_",
1582            "/_/health",
1583            "/_/ready/deep",
1584        ] {
1585            let refused = refusal(&[given]);
1586            assert!(refused.contains("serves"), "{given:?}: {refused}");
1587        }
1588    }
1589
1590    #[test]
1591    fn a_prefix_sharing_only_characters_with_a_reserved_one_is_accepted() {
1592        let nests = mount_points(routes(&[
1593            "/apis",
1594            "/v1/other",
1595            "/administration",
1596            "/_/hooks",
1597            "/__",
1598        ]))
1599        .unwrap();
1600        assert_eq!(nests.len(), 5);
1601    }
1602
1603    #[test]
1604    fn two_prefixes_that_overlap_are_refused() {
1605        assert!(refusal(&["/hooks", "hooks/"]).contains("nested earlier"));
1606        assert!(refusal(&["/hooks", "/hooks/stripe"]).contains("nested earlier"));
1607        assert!(refusal(&["/hooks/stripe", "/hooks"]).contains("nested earlier"));
1608        assert!(refusal(&["/_/hooks", "/_/hooks"]).contains("nested earlier"));
1609    }
1610
1611    /// Whatever `mount_points` accepts, the router takes without panicking.
1612    /// The built-in routes are laid out as `assemble` lays them out, with
1613    /// every feature's paths present whether or not the feature is compiled
1614    /// in, and each nested router serving something, since an empty one
1615    /// claims no path.
1616    #[test]
1617    fn what_is_accepted_does_not_panic_the_router() {
1618        let stub = || Router::new().route("/", any(|| async {}));
1619        let nests = mount_points(
1620            [
1621                "/_/hooks",
1622                "/_/jobs/nightly",
1623                "/hooks",
1624                "/v1/other",
1625                "/apis",
1626            ]
1627            .into_iter()
1628            .map(|prefix| (prefix.to_string(), stub()))
1629            .collect(),
1630        )
1631        .unwrap();
1632        let mut app: Router<Arc<AppState>> = Router::new()
1633            .nest("/api", Router::new().route("/{*path}", any(|| async {})))
1634            .route("/healthz", any(|| async {}))
1635            .route("/v1/version", any(|| async {}))
1636            .nest(CUSTOM_PREFIX, custom::custom_router());
1637        for (prefix, router) in nests {
1638            app = app.nest(&prefix, router);
1639        }
1640        let _ = app
1641            .nest("/admin", stub())
1642            .nest("/v1/graphql", stub())
1643            .nest("/v1alpha1/graphql", stub())
1644            .nest("/api/graphql", stub())
1645            .route("/", any(|| async {}))
1646            .fallback(|| async {});
1647    }
1648}