cja/server/
mod.rs

1//! HTTP server built on Axum with cookies, sessions, and zero-downtime reload.
2//!
3//! # Routes
4//!
5//! Build routes using standard Axum patterns and pass them to [`run_server`]:
6//!
7//! ```rust,ignore
8//! fn routes(app_state: MyAppState) -> axum::Router {
9//!     axum::Router::new()
10//!         .route("/", axum::routing::get(index))
11//!         .route("/api/health", axum::routing::get(health))
12//!         .with_state(app_state)
13//! }
14//! ```
15//!
16//! # Sessions
17//!
18//! Implement [`session::AppSession`] for your session type, then use
19//! [`session::Session<T>`](session::Session) as an Axum extractor:
20//!
21//! ```rust,ignore
22//! async fn index(Session(session): Session<MySession>) -> impl IntoResponse {
23//!     html! {
24//!         p { "Session: " (session.inner().session_id) }
25//!     }
26//! }
27//! ```
28//!
29//! Sessions are automatically created on first access and persisted via encrypted cookies.
30//!
31//! # Zero-Downtime Reload
32//!
33//! ```bash
34//! systemfd --no-pid -s http::3000 -- cargo watch -x 'run -p my-app'
35//! ```
36//!
37//! CJA checks for the `LISTEN_FDS` environment variable set by `systemfd` and reuses
38//! the existing socket when present, allowing restarts without dropping connections.
39
40use axum::{extract::Request, response::Response};
41use color_eyre::eyre::WrapErr;
42use listenfd::ListenFd;
43use std::{convert::Infallible, error::Error, future::Future, net::SocketAddr};
44use tokio::net::TcpListener;
45use tower_cookies::CookieManagerLayer;
46use tower_service::Service;
47
48pub mod cookies {
49    mod cookie_key;
50    pub use cookie_key::CookieKey;
51
52    mod cookie_jar;
53    pub use cookie_jar::CookieJar;
54
55    pub use tower_cookies::Cookie;
56
57    pub use tower_cookies::cookie::SameSite;
58}
59
60pub mod page;
61
62pub mod session;
63
64pub mod trace;
65
66pub async fn run_server<AS: Clone + Sync + Send + 'static, S, E>(
67    routes: axum::Router<AS>,
68) -> color_eyre::Result<()>
69where
70    for<'a> axum::Router<AS>: tower_service::Service<
71            axum::serve::IncomingStream<'a, tokio::net::TcpListener>,
72            Error = Infallible,
73            Response = S,
74        > + Send
75        + Clone,
76    S: Service<Request, Response = Response, Error = Infallible> + Clone + Send + 'static,
77    S::Future: Send,
78    axum::serve::Serve<tokio::net::TcpListener, axum::Router<AS>, S>:
79        IntoFuture<Output = Result<(), E>>,
80    E: Error + Send + Sync + 'static,
81{
82    let app = prepare_routes(routes);
83    let listener = bind_listener().await?;
84    axum::serve(listener, app)
85        .await
86        .wrap_err("Failed to run server")
87}
88
89/// Run the standard cja HTTP server until `shutdown` resolves, then stop
90/// accepting connections and finish in-flight requests. Preserves `PORT`,
91/// systemfd listeners, request tracing and cookie middleware.
92///
93/// The caller should bound the drain if a request can run indefinitely.
94pub async fn run_server_until(
95    routes: axum::Router,
96    shutdown: impl Future<Output = ()> + Send + 'static,
97) -> color_eyre::Result<()> {
98    serve_until(bind_listener().await?, routes, shutdown).await
99}
100
101/// Serve an already-bound listener with cja middleware and graceful shutdown.
102/// This is useful when the application owns its bind address or port.
103pub async fn serve_until(
104    listener: TcpListener,
105    routes: axum::Router,
106    shutdown: impl Future<Output = ()> + Send + 'static,
107) -> color_eyre::Result<()> {
108    axum::serve(listener, prepare_routes(routes))
109        .with_graceful_shutdown(shutdown)
110        .await
111        .wrap_err("Failed to run server")
112}
113
114fn prepare_routes<AS: Clone + Send + Sync + 'static>(routes: axum::Router<AS>) -> axum::Router<AS> {
115    let tracer = trace::Tracer;
116    let trace_layer = tower_http::trace::TraceLayer::new_for_http()
117        .make_span_with(tracer)
118        .on_response(tracer);
119
120    routes.layer(trace_layer).layer(CookieManagerLayer::new())
121}
122
123async fn bind_listener() -> color_eyre::Result<TcpListener> {
124    let port = std::env::var("PORT").unwrap_or_else(|_| "3000".to_string());
125    let port: u16 = port.parse()?;
126    let addr = SocketAddr::from(([0, 0, 0, 0], port));
127
128    // Check if we're being run under systemfd (LISTEN_FDS will be set)
129    let listener = if let Some(fd_listener) = ListenFd::from_env().take_tcp_listener(0)? {
130        // If systemfd is being used, we'll get a listener from fd
131        let socket_addr = fd_listener.local_addr()?;
132
133        tracing::info!("Zero-downtime reloading enabled");
134        tracing::info!(
135            "Using listener passed from systemfd on address {}",
136            socket_addr
137        );
138
139        // Convert the std TcpListener to a tokio one
140        TcpListener::from_std(fd_listener)?
141    } else {
142        // Otherwise, create our own listener
143        tracing::info!("Starting server on port {}", port);
144        TcpListener::bind(&addr)
145            .await
146            .wrap_err("Failed to open port")?
147    };
148
149    let addr = listener.local_addr()?;
150    tracing::info!("Listening on {}", addr);
151
152    Ok(listener)
153}
cja/server/mod.rs

cja/server/
mod.rs
