edda_server/
lib.rs

1//! What every edda service shares: logging, opening its
2//! database, finding the auth service's key, serving (with the CORS setting
3//! and the limits), error responses, and the conventions of their `OpenAPI`
4//! descriptions.
5
6pub mod admin;
7pub mod limits;
8pub mod measure;
9pub mod observe;
10pub mod openapi;
11#[cfg(feature = "testing")]
12pub mod testing;
13
14pub use limits::Limits;
15
16use std::{error::Error, fmt::Display, time::Duration};
17
18use axum::{
19    Json, Router,
20    http::{HeaderValue, StatusCode, header},
21    response::{IntoResponse, Response},
22};
23use edda_auth::PublicKey;
24use edda_client::auth::AuthClient;
25use serde_json::json;
26use sqlx::{
27    Connection, PgConnection, PgPool, Postgres,
28    migrate::{MigrateDatabase, Migrator},
29    postgres::PgPoolOptions,
30};
31use tower_http::cors::{AllowOrigin, Any, CorsLayer};
32use tracing_subscriber::EnvFilter;
33
34/// `GET /health`: answers `ok` while the service runs.
35#[utoipa::path(
36    get,
37    path = "/health",
38    tag = "health",
39    security(()),
40    responses((status = 200, description = "The service is running.", body = String, example = "ok"))
41)]
42#[expect(clippy::unused_async, reason = "axum handlers are async")]
43pub async fn health() -> &'static str {
44    "ok"
45}
46
47/// How long `/ready` waits for the database before saying it isn't.
48const READY_TIMEOUT: Duration = Duration::from_secs(2);
49
50/// `GET /ready`, for whatever runs the service (Kubernetes, compose): `200`
51/// while it can reach its database, `503` while it can't, so it gets no
52/// requests until it can again. `/health` only says it's running.
53pub fn readiness(pool: PgPool) -> Router {
54    Router::new()
55        .route("/ready", axum::routing::get(ready))
56        .with_state(pool)
57}
58
59/// `GET /ready`: see [`readiness`].
60#[utoipa::path(
61    get,
62    path = "/ready",
63    tag = "health",
64    security(()),
65    responses(
66        (status = 200, description = "The service can reach its database.", body = String, example = "ready"),
67        (status = 503, description = "It can't, for now.", body = openapi::ErrorBody),
68    )
69)]
70pub async fn ready(axum::extract::State(pool): axum::extract::State<PgPool>) -> Response {
71    let checked = tokio::time::timeout(READY_TIMEOUT, sqlx::query("SELECT 1").execute(&pool)).await;
72    match checked {
73        Ok(Ok(_)) => "ready".into_response(),
74        Ok(Err(e)) => {
75            tracing::warn!(error = %e, "not ready: the database refuses");
76            error_response(
77                StatusCode::SERVICE_UNAVAILABLE,
78                &"the database can't be reached",
79            )
80        }
81        Err(_) => {
82            tracing::warn!("not ready: the database doesn't answer");
83            error_response(
84                StatusCode::SERVICE_UNAVAILABLE,
85                &"the database can't be reached",
86            )
87        }
88    }
89}
90
91/// Logs to stderr: `crate_name`'s own messages, this crate's (serving,
92/// internal errors) and each request, at info, unless `RUST_LOG` says
93/// otherwise. `LOG_FORMAT=json` writes a JSON object a line, for a log
94/// collector, and `OTEL_EXPORTER_OTLP_ENDPOINT` sends traces to an
95/// OpenTelemetry collector; see [`observe`].
96pub fn init_logging(crate_name: &str) {
97    let filter = EnvFilter::try_from_default_env()
98        .unwrap_or_else(|_| format!("{crate_name}=info,edda_server=info,tower_http=info").into());
99    let json = std::env::var("LOG_FORMAT").is_ok_and(|f| f == "json");
100    let traces = observe::traces(crate_name).unwrap_or_else(|e| {
101        eprintln!("not sending traces: {e}");
102        None
103    });
104    tracing::subscriber::set_global_default(observe::subscriber(
105        json,
106        filter,
107        std::io::stderr,
108        traces,
109    ))
110    .expect("logging is set up once");
111}
112
113/// A connection pool to the service's Postgres database at `url`, after
114/// creating the database if there's none yet and bringing it up to date
115/// with `migrations`. Those two go through [`direct_database_url`], as
116/// they need a session of their own (sqlx serialises migrations with an
117/// advisory lock, so instances may start at once).
118pub async fn open_database(url: &str, migrations: &Migrator) -> Result<PgPool, sqlx::Error> {
119    let direct = direct_database_url(url);
120    if !Postgres::database_exists(&direct).await? {
121        // Another instance may be creating it at the same moment.
122        if let Err(e) = Postgres::create_database(&direct).await
123            && !Postgres::database_exists(&direct).await?
124        {
125            return Err(e);
126        }
127    }
128    let mut session = PgConnection::connect(&direct).await?;
129    migrations.run(&mut session).await?;
130    session.close().await?;
131    PgPoolOptions::new().connect(url).await
132}
133
134/// Where to reach the database for what needs a session of its own:
135/// migrations, and the combat service's `LISTEN`. `DATABASE_DIRECT_URL`
136/// when `url` goes through a pooler that hands out connections per
137/// transaction (`PgBouncer`'s transaction mode), otherwise `url` itself.
138pub fn direct_database_url(url: &str) -> String {
139    direct_or(url, std::env::var("DATABASE_DIRECT_URL").ok())
140}
141
142fn direct_or(url: &str, direct: Option<String>) -> String {
143    direct
144        .filter(|d| !d.is_empty())
145        .unwrap_or_else(|| url.to_owned())
146}
147
148/// The key the auth service signs tokens with: `AUTH_PUBLIC_KEY` if set,
149/// otherwise fetched from the auth service at `auth_service` (waiting a
150/// while for it to start). Set the key explicitly unless the network
151/// between the services is trusted.
152pub async fn auth_key(auth_service: &str) -> Result<PublicKey, Box<dyn Error>> {
153    if let Ok(key) = std::env::var("AUTH_PUBLIC_KEY") {
154        return Ok(key.parse()?);
155    }
156    let key = AuthClient::new(auth_service)
157        .wait_for_server_key(Duration::from_secs(30))
158        .await
159        .map_err(|e| {
160            format!(
161                "could not fetch the auth service's key from {auth_service} ({e}); \
162                 start it or set AUTH_PUBLIC_KEY"
163            )
164        })?;
165    tracing::info!(%key, "using the auth service's token key");
166    Ok(key)
167}
168
169/// `AUTH_SERVICE_URL`, or the auth service on this machine.
170pub fn auth_service_url() -> String {
171    std::env::var("AUTH_SERVICE_URL").unwrap_or_else(|_| "http://127.0.0.1:3002".to_string())
172}
173
174/// Serves `app` on `bind` until asked to stop, within `limits`, answering
175/// web pages from the origins in `CORS_ORIGINS`; and `admin` on
176/// `ADMIN_ADDR` and metrics on `METRICS_ADDR`, if set.
177pub async fn serve(
178    app: Router,
179    admin: Router,
180    bind: &str,
181    name: &str,
182    limits: &Limits,
183) -> Result<(), Box<dyn Error>> {
184    if let Some(config) = admin::Admin::from_env()? {
185        let listener = tokio::net::TcpListener::bind(&config.address).await?;
186        tracing::info!(address = %config.address, "{name}'s admin API listening");
187        let admin = observe::requests(config.guard(admin), false);
188        tokio::spawn(async move {
189            if let Err(e) = axum::serve(listener, admin).await {
190                tracing::error!(error = %e, "serving the admin API failed");
191            }
192        });
193    }
194    let app = match std::env::var("METRICS_ADDR") {
195        Ok(address) => {
196            let measured = measure::endpoint(measure::install(name)?);
197            let listener = tokio::net::TcpListener::bind(&address).await?;
198            tracing::info!(%address, "{name}'s metrics at /metrics");
199            tokio::spawn(async move {
200                if let Err(e) = axum::serve(listener, measured).await {
201                    tracing::error!(error = %e, "serving metrics failed");
202                }
203            });
204            measure::requests(app)
205        }
206        Err(_) => app,
207    };
208    let app = observe::requests(limits.apply(app.layer(cors()?)), limits.trust_proxy);
209    let listener = tokio::net::TcpListener::bind(bind).await?;
210    tracing::info!(%bind, limits = limits.enabled(), "{name} listening");
211    // Each request knows its client's address, for the limits.
212    let app = app.into_make_service_with_connect_info::<std::net::SocketAddr>();
213    axum::serve(listener, app)
214        .with_graceful_shutdown(stopping())
215        .await?;
216    tracing::info!("{name} stopped");
217    observe::flush();
218    Ok(())
219}
220
221/// Resolves when the service is asked to stop: Ctrl-C, or SIGTERM, which
222/// is how Kubernetes and compose ask. Requests in progress then finish;
223/// open WebSockets close, and their clients reconnect elsewhere.
224async fn stopping() {
225    let interrupted = async {
226        let _ = tokio::signal::ctrl_c().await;
227    };
228    #[cfg(unix)]
229    let terminated = async {
230        use tokio::signal::unix::{SignalKind, signal};
231        match signal(SignalKind::terminate()) {
232            Ok(mut terminate) => {
233                terminate.recv().await;
234            }
235            Err(e) => {
236                tracing::warn!(error = %e, "can't listen for SIGTERM");
237                std::future::pending::<()>().await;
238            }
239        }
240    };
241    #[cfg(not(unix))]
242    let terminated = std::future::pending::<()>();
243    tokio::select! {
244        () = interrupted => {}
245        () = terminated => {}
246    }
247    tracing::info!("stopping");
248}
249
250/// What an error response says: the error itself, unless it's internal
251/// (a 500), which is logged and not shown.
252pub fn error_message(status: StatusCode, error: &dyn Display) -> String {
253    if status == StatusCode::INTERNAL_SERVER_ERROR {
254        tracing::error!(%error, "internal error");
255        "internal server error".to_string()
256    } else {
257        error.to_string()
258    }
259}
260
261/// An error response: `{"error": "..."}` with `status` (see
262/// [`error_message`]).
263pub fn error_response(status: StatusCode, error: &dyn Display) -> Response {
264    (
265        status,
266        Json(json!({ "error": error_message(status, error) })),
267    )
268        .into_response()
269}
270
271/// Which web pages may call the service from a browser, from
272/// `CORS_ORIGINS`: a comma-separated list of origins (such as
273/// `https://dnd.example.com`), or `*` for any (for development). Unset,
274/// none may: `edda` doesn't need it, and neither does a web page
275/// served from the same address as the services (behind the proxy).
276fn cors() -> Result<CorsLayer, String> {
277    let setting = std::env::var("CORS_ORIGINS").unwrap_or_default();
278    cors_for(&setting)
279}
280
281fn cors_for(setting: &str) -> Result<CorsLayer, String> {
282    let origins: Vec<&str> = setting
283        .split(',')
284        .map(str::trim)
285        .filter(|o| !o.is_empty())
286        .collect();
287    if origins.is_empty() {
288        return Ok(CorsLayer::new());
289    }
290    let allowed = if origins == ["*"] {
291        AllowOrigin::any()
292    } else {
293        let parsed: Result<Vec<HeaderValue>, _> = origins
294            .iter()
295            .map(|o| {
296                HeaderValue::from_str(o).map_err(|_| format!("CORS_ORIGINS: {o:?} isn't an origin"))
297            })
298            .collect();
299        AllowOrigin::list(parsed?)
300    };
301    // Tokens travel in the Authorization header, not cookies, so no
302    // credentials are needed.
303    Ok(CorsLayer::new()
304        .allow_origin(allowed)
305        .allow_methods(Any)
306        .allow_headers(Any)
307        .expose_headers([header::ETAG, header::LOCATION]))
308}
309
310#[cfg(test)]
311mod tests {
312    use axum::{body::Body, http::Request};
313    use sqlx::postgres::PgPoolOptions;
314    use tower::ServiceExt;
315
316    use super::*;
317
318    #[tokio::test]
319    async fn it_is_not_ready_while_its_database_cant_be_reached() {
320        // Nothing listens on port 1.
321        let pool = PgPoolOptions::new()
322            .acquire_timeout(Duration::from_millis(200))
323            .connect_lazy("postgres://postgres@127.0.0.1:1/nothing")
324            .unwrap();
325        let res = readiness(pool)
326            .oneshot(Request::get("/ready").body(Body::empty()).unwrap())
327            .await
328            .unwrap();
329        assert_eq!(res.status(), StatusCode::SERVICE_UNAVAILABLE);
330    }
331
332    #[test]
333    fn a_direct_url_is_the_pooled_one_unless_given() {
334        let pooled = "postgres://pgbouncer:6432/combats";
335        assert_eq!(direct_or(pooled, None), pooled);
336        assert_eq!(direct_or(pooled, Some(String::new())), pooled);
337        let direct = "postgres://postgres:5432/combats";
338        assert_eq!(direct_or(pooled, Some(direct.into())), direct);
339    }
340
341    #[test]
342    fn origins_are_a_list_or_any() {
343        assert!(cors_for("").is_ok());
344        assert!(cors_for("https://dnd.example.com, http://192.168.1.20:8080").is_ok());
345        assert!(cors_for("*").is_ok());
346        assert!(cors_for("https://bad\norigin").is_err());
347    }
348}
edda_server/lib.rs

edda_server/
lib.rs
