1use crate::dep::{AnyArc, DepEnv, DepFactory, DepResolver, TaskContext};
5use crate::error::{Error, Result};
6use crate::extract::{BodyLane, RequestCtx};
7use crate::handler::BoxHandlerFn;
8use crate::middleware::{Middleware, Next};
9use crate::module::{FlatRoute, Module};
10use crate::response::{IntoResponse, Response};
11use crate::router::{Endpoint, MethodRouter, RouteMatch, Trie};
12use crate::serve;
13#[cfg(test)]
14use crate::serve::is_transient_accept_error;
15#[cfg(test)]
16use bytes::Bytes;
17use std::any::TypeId;
18use std::collections::HashMap;
19use std::future::Future;
20use std::pin::Pin;
21use std::sync::Arc;
22
23pub(crate) type BackgroundFactory = Box<
27 dyn FnOnce(
28 TaskContext,
29 tokio::sync::watch::Receiver<bool>,
30 ) -> Pin<Box<dyn Future<Output = ()> + Send>>
31 + Send,
32>;
33
34pub type ErrorBodyMapper =
42 dyn Fn(http::StatusCode, &str, &str) -> Option<serde_json::Value> + Send + Sync;
43
44pub struct App {
47 routes: Vec<(String, MethodRouter)>,
48 mounts: Vec<(String, Module)>,
49 env: DepEnv,
50 middleware: Vec<Arc<dyn Middleware>>,
51 security_headers: bool,
52 cors: Option<std::sync::Arc<crate::cors::CorsConfig>>,
53 handler_timeout: std::time::Duration,
54 body_read_timeout: std::time::Duration,
55 write_stall_timeout: std::time::Duration,
56 error_mapper: Option<Arc<ErrorBodyMapper>>,
59 background: Vec<(&'static str, BackgroundFactory)>,
62}
63
64impl Default for App {
65 fn default() -> Self {
66 let mut env = DepEnv::default();
71 env.insert_value(crate::clock::Clock::system());
72 Self {
73 routes: Vec::new(),
74 mounts: Vec::new(),
75 env,
76 middleware: Vec::new(),
77 security_headers: true,
78 cors: None,
79 handler_timeout: std::time::Duration::from_secs(30),
80 body_read_timeout: std::time::Duration::from_secs(30),
81 write_stall_timeout: std::time::Duration::from_secs(30),
82 error_mapper: None,
83 background: Vec::new(),
84 }
85 }
86}
87
88pub trait Extension {
91 fn register(self, app: App) -> App;
92}
93
94impl App {
95 pub fn new() -> Self {
96 Self::default()
97 }
98
99 pub fn extend<E: Extension>(self, extension: E) -> App {
101 extension.register(self)
102 }
103
104 pub fn security_headers(mut self, on: bool) -> Self {
107 self.security_headers = on;
108 self
109 }
110
111 pub fn map_error_body<F>(mut self, f: F) -> Self
118 where
119 F: Fn(http::StatusCode, &str, &str) -> Option<serde_json::Value> + Send + Sync + 'static,
120 {
121 self.error_mapper = Some(Arc::new(f));
122 self
123 }
124
125 pub fn cors(mut self, config: crate::cors::CorsConfig) -> Self {
130 self.cors = Some(std::sync::Arc::new(config));
131 self
132 }
133
134 pub fn handler_timeout(mut self, budget: std::time::Duration) -> Self {
137 self.handler_timeout = budget;
138 self
139 }
140
141 pub fn body_read_timeout(mut self, budget: std::time::Duration) -> Self {
143 self.body_read_timeout = budget;
144 self
145 }
146
147 pub fn write_stall_timeout(mut self, budget: std::time::Duration) -> Self {
151 self.write_stall_timeout = budget;
152 self
153 }
154
155 pub fn route(mut self, path: &str, methods: MethodRouter) -> Self {
157 self.routes.push((path.to_string(), methods));
158 self
159 }
160
161 pub fn mount(mut self, prefix: &str, module: Module) -> Self {
163 self.mounts.push((prefix.to_string(), module));
164 self
165 }
166
167 pub fn provide<T: Send + Sync + 'static>(mut self, value: T) -> Self {
169 self.env.insert_value(value);
170 self
171 }
172
173 pub fn provide_dep<F, Args, T>(mut self, factory: F) -> Self
175 where
176 F: DepFactory<Args, T>,
177 T: Send + Sync + 'static,
178 {
179 self.env.insert_factory(factory);
180 self
181 }
182
183 pub fn middleware<M: Middleware>(mut self, mw: M) -> Self {
185 self.middleware.push(Arc::new(mw));
186 self
187 }
188
189 pub fn on_serve<F, Fut>(mut self, name: &'static str, f: F) -> App
202 where
203 F: FnOnce(TaskContext, tokio::sync::watch::Receiver<bool>) -> Fut + Send + 'static,
204 Fut: Future<Output = ()> + Send + 'static,
205 {
206 let factory: BackgroundFactory = Box::new(move |ctx, shutdown| Box::pin(f(ctx, shutdown)));
207 self.background.push((name, factory));
208 self
209 }
210
211 pub(crate) fn take_background(&mut self) -> Vec<(&'static str, BackgroundFactory)> {
215 std::mem::take(&mut self.background)
216 }
217
218 pub fn build(self) -> Result<BuiltApp> {
221 if let Some(c) = &self.cors {
222 c.validate()?;
223 }
224 let mut trie = Trie::default();
225 let app_env = Arc::new(self.env.clone());
226 let app_mw: Arc<[Arc<dyn Middleware>]> = Arc::from(self.middleware.clone());
227
228 for (path, methods) in self.routes {
229 let body_limit = methods.body_limit;
230 insert_flat(
231 &mut trie,
232 FlatRoute {
233 path,
234 methods,
235 env: app_env.clone(),
236 middleware: app_mw.clone(),
237 body_limit,
238 },
239 )?;
240 }
241 for (prefix, module) in self.mounts {
242 for flat in module.flatten(&prefix, &self.env, &self.middleware) {
243 insert_flat(&mut trie, flat)?;
244 }
245 }
246 Ok(BuiltApp {
247 trie,
248 app_env,
249 overrides: Arc::new(HashMap::new()),
250 security_headers: self.security_headers,
251 cors: self.cors.clone(),
252 handler_timeout: self.handler_timeout,
253 body_read_timeout: self.body_read_timeout,
254 write_stall_timeout: self.write_stall_timeout,
255 error_mapper: self.error_mapper.clone(),
256 })
257 }
258
259 pub async fn serve(self) -> Result<()> {
263 let addr = std::env::var("JERRYCAN_ADDR").unwrap_or_else(|_| "127.0.0.1:8000".to_string());
264 let listener = tokio::net::TcpListener::bind(&addr)
265 .await
266 .map_err(|e| Error::internal(format!("failed to bind {addr}: {e}")))?;
267 self.serve_with_shutdown(listener, serve::shutdown_signal())
268 .await
269 }
270
271 pub async fn serve_with(self, listener: tokio::net::TcpListener) -> Result<()> {
273 self.serve_with_shutdown(listener, std::future::pending())
274 .await
275 }
276
277 pub async fn serve_with_shutdown(
280 self,
281 listener: tokio::net::TcpListener,
282 shutdown: impl std::future::Future<Output = ()> + Send,
283 ) -> Result<()> {
284 serve::run_with_shutdown(self, listener, shutdown).await
285 }
286}
287
288fn insert_flat(trie: &mut Trie, flat: FlatRoute) -> Result<()> {
289 let stream_body = flat.methods.stream_body;
290 let mut methods = HashMap::new();
291 for (m, h) in flat.methods.handlers {
292 if methods.insert(m.clone(), h).is_some() {
293 return Err(Error::internal(format!(
294 "duplicate method {m} for `{}`",
295 flat.path
296 )));
297 }
298 }
299 trie.insert(
300 &flat.path,
301 Endpoint {
302 methods,
303 env: flat.env,
304 middleware: flat.middleware,
305 body_limit: flat.body_limit,
306 stream_body,
307 },
308 )
309}
310
311pub struct BuiltApp {
313 pub(crate) trie: Trie,
314 pub(crate) app_env: Arc<DepEnv>,
318 pub(crate) overrides: Arc<HashMap<TypeId, AnyArc>>,
319 pub(crate) security_headers: bool,
320 pub(crate) cors: Option<std::sync::Arc<crate::cors::CorsConfig>>,
323 pub(crate) handler_timeout: std::time::Duration,
324 pub(crate) body_read_timeout: std::time::Duration,
325 pub(crate) write_stall_timeout: std::time::Duration,
326 pub(crate) error_mapper: Option<Arc<ErrorBodyMapper>>,
329}
330
331impl std::fmt::Debug for BuiltApp {
334 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
335 f.debug_struct("BuiltApp").finish_non_exhaustive()
336 }
337}
338
339pub(crate) const BODY_LIMIT: usize = 1024 * 1024; pub(crate) enum Policy {
348 Route { limit: usize, stream: bool },
352 Reject(Response),
355}
356
357pub(crate) fn apply_security_headers(res: &mut Response) {
359 const DEFAULTS: [(&str, &str); 5] = [
360 ("x-content-type-options", "nosniff"),
361 ("x-frame-options", "DENY"),
362 ("referrer-policy", "no-referrer"),
363 ("content-security-policy", "default-src 'none'"),
364 ("cache-control", "no-store"),
365 ];
366 for (name, value) in DEFAULTS {
367 let header_name = http::HeaderName::from_static(name);
368 if !res.headers().contains_key(&header_name) {
369 res.headers_mut()
370 .insert(header_name, http::HeaderValue::from_static(value));
371 }
372 }
373}
374
375impl BuiltApp {
376 pub fn task_context(&self) -> crate::dep::TaskContext {
384 crate::dep::TaskContext::new(DepResolver::new(
385 self.app_env.clone(),
386 self.overrides.clone(),
387 ))
388 }
389
390 pub(crate) fn apply_error_mapper(&self, response: &mut Response) {
398 let Some(mapper) = &self.error_mapper else {
399 return;
400 };
401 let Some(parts) = response
402 .extensions()
403 .get::<crate::response::ErrorParts>()
404 .cloned()
405 else {
406 return;
407 };
408 if let Some(body) = mapper(response.status(), parts.code, &parts.message) {
409 let bytes = serde_json::to_vec(&body).unwrap_or_default();
410 *response.body_mut() = crate::response::JcBody::full(bytes);
411 }
412 }
413
414 pub(crate) fn route_policy(&self, parts: &http::request::Parts) -> Policy {
426 let path = parts.uri.path();
427 if let Some(config) = &self.cors
432 && crate::cors::is_preflight(parts)
433 {
434 let origin = parts
435 .headers
436 .get(http::header::ORIGIN)
437 .and_then(|v| v.to_str().ok())
438 .unwrap_or("");
439 if config.allows_origin(origin)
440 && let Some(methods) = self.trie.methods_for(path)
441 {
442 let acrh = parts
443 .headers
444 .get(http::header::ACCESS_CONTROL_REQUEST_HEADERS)
445 .and_then(|v| v.to_str().ok());
446 return Policy::Reject(crate::cors::preflight_response(
447 config, origin, acrh, &methods,
448 ));
449 }
450 let mut r = http::Response::new(crate::response::JcBody::empty());
453 *r.status_mut() = http::StatusCode::NO_CONTENT;
454 return Policy::Reject(r);
455 }
456 let reject = |response: Response| -> Policy {
457 let mut response = response;
458 self.apply_error_mapper(&mut response);
461 if self.security_headers {
462 apply_security_headers(&mut response);
463 }
464 if let Some(config) = &self.cors {
469 crate::cors::apply_cors(
470 &mut response,
471 parts.headers.get(http::header::ORIGIN),
472 config,
473 );
474 }
475 Policy::Reject(response)
476 };
477 match self.trie.find(path, &parts.method) {
478 RouteMatch::Found { endpoint, .. } => Policy::Route {
479 limit: endpoint.body_limit.unwrap_or(BODY_LIMIT),
480 stream: endpoint.stream_body,
481 },
482 RouteMatch::NotFound => reject(Error::not_found().into_response()),
483 RouteMatch::MethodMissing => reject(Error::method_not_allowed().into_response()),
484 RouteMatch::Malformed => {
485 reject(Error::bad_request("malformed percent-encoding in path").into_response())
486 }
487 }
488 }
489
490 pub(crate) async fn dispatch(&self, parts: http::request::Parts, lane: BodyLane) -> Response {
495 let origin = parts.headers.get(http::header::ORIGIN).cloned();
499 let mut response = self.dispatch_inner(parts, lane).await;
500 self.apply_error_mapper(&mut response);
503 if self.security_headers {
504 apply_security_headers(&mut response);
505 }
506 if let Some(config) = &self.cors {
507 crate::cors::apply_cors(&mut response, origin.as_ref(), config);
508 }
509 response
510 }
511
512 async fn dispatch_inner(&self, parts: http::request::Parts, lane: BodyLane) -> Response {
513 let method = parts.method.clone();
514 let path = parts.uri.path().to_string();
515 match self.trie.find(&path, &method) {
516 RouteMatch::NotFound => Error::not_found().into_response(),
517 RouteMatch::MethodMissing => Error::method_not_allowed().into_response(),
518 RouteMatch::Malformed => {
519 Error::bad_request("malformed percent-encoding in path").into_response()
520 }
521 RouteMatch::Found { endpoint, params } => {
522 let mut ctx = RequestCtx::with_lane(
523 parts,
524 lane,
525 DepResolver::new(endpoint.env.clone(), self.overrides.clone()),
526 );
527 ctx.params = params;
528 let handler: &BoxHandlerFn = endpoint
529 .methods
530 .get(&method)
531 .expect("find() checked the method");
532 let run = Next {
533 chain: &endpoint.middleware,
534 endpoint: handler,
535 }
536 .run(&mut ctx);
537 match tokio::time::timeout(self.handler_timeout, run).await {
538 Ok(response) => response,
539 Err(_) => Error::handler_timeout().into_response(),
540 }
541 }
542 }
543 }
544}
545
546#[cfg(test)]
547mod tests {
548 use super::*;
549 use crate::response::Json;
550 use crate::router::get;
551 use crate::{Dep, Path};
552 use std::sync::Mutex;
553
554 #[derive(Default)]
555 struct Store {
556 items: Mutex<Vec<String>>,
557 }
558
559 async fn list(store: Dep<Store>) -> Json<Vec<String>> {
560 Json(store.items.lock().unwrap().clone())
561 }
562
563 async fn create(store: Dep<Store>, Json(item): Json<String>) -> crate::Result<Json<usize>> {
564 let mut items = store.items.lock().unwrap();
565 items.push(item);
566 Ok(Json(items.len()))
567 }
568
569 async fn show(store: Dep<Store>, Path(ix): Path<usize>) -> crate::Result<Json<String>> {
570 store
571 .items
572 .lock()
573 .unwrap()
574 .get(ix)
575 .cloned()
576 .map(Json)
577 .ok_or_else(Error::not_found)
578 }
579
580 fn crud_app() -> App {
581 App::new().provide(Store::default()).mount(
582 "/todos",
583 Module::new("todos")
584 .route("/", get(list).post(create))
585 .route("/{ix}", get(show)),
586 )
587 }
588
589 async fn dispatch(built: &BuiltApp, method: http::Method, path: &str, body: &str) -> Response {
590 let req = http::Request::builder()
591 .method(method)
592 .uri(path)
593 .body(())
594 .unwrap();
595 let (parts, ()) = req.into_parts();
596 built
597 .dispatch(parts, BodyLane::Buffered(Bytes::from(body.to_string())))
598 .await
599 }
600
601 #[tokio::test]
602 async fn crud_round_trip_in_process() {
603 let built = crud_app().build().unwrap();
604 let r = dispatch(&built, http::Method::POST, "/todos/", r#""write spike""#).await;
605 assert_eq!(r.status(), http::StatusCode::OK);
606 let r = dispatch(&built, http::Method::GET, "/todos/0", "").await;
607 assert_eq!(r.status(), http::StatusCode::OK);
608 let r = dispatch(&built, http::Method::GET, "/todos/9", "").await;
609 assert_eq!(r.status(), http::StatusCode::NOT_FOUND);
610 let r = dispatch(&built, http::Method::PATCH, "/todos/", "").await;
611 assert_eq!(r.status(), http::StatusCode::METHOD_NOT_ALLOWED);
612 let r = dispatch(&built, http::Method::GET, "/nope", "").await;
613 assert_eq!(r.status(), http::StatusCode::NOT_FOUND);
614 }
615
616 async fn body_string(r: Response) -> String {
617 use http_body_util::BodyExt;
618 let bytes = r.into_body().collect().await.unwrap().to_bytes();
619 String::from_utf8(bytes.to_vec()).unwrap()
620 }
621
622 #[tokio::test]
629 async fn registered_error_mapper_reshapes_framework_error_bodies() {
630 async fn guarded() -> crate::Result<&'static str> {
631 Err(Error::unauthorized()) }
633 let secure = || Module::new("secure").route("/", get(guarded));
634
635 let mapped = App::new()
636 .map_error_body(|_status, code, message| {
637 Some(serde_json::json!({ "error": { "code": code, "message": message } }))
638 })
639 .mount("/secure", secure())
640 .build()
641 .unwrap();
642 let r = dispatch(&mapped, http::Method::GET, "/secure/", "").await;
644 assert_eq!(r.status(), http::StatusCode::UNAUTHORIZED);
645 assert_eq!(
646 body_string(r).await,
647 r#"{"error":{"code":"JC0401","message":"authentication required"}}"#
648 );
649 let r = dispatch(&mapped, http::Method::GET, "/nope", "").await;
651 let body = body_string(r).await;
652 assert!(
653 body.contains(r#""error""#) && body.contains("JC0404"),
654 "routing 404 must be reshaped: {body}"
655 );
656
657 let plain = App::new().mount("/secure", secure()).build().unwrap();
659 let r = dispatch(&plain, http::Method::GET, "/secure/", "").await;
660 assert_eq!(
661 body_string(r).await,
662 r#"{"code":"JC0401","message":"authentication required"}"#
663 );
664 }
665
666 #[tokio::test]
669 async fn error_mapper_returning_none_preserves_the_default_body_and_details() {
670 async fn detailed() -> crate::Result<&'static str> {
671 Err(Error::unprocessable("bad")
672 .with_details(serde_json::json!([{ "field": "x", "message": "required" }])))
673 }
674 let built = App::new()
675 .map_error_body(|_s, _c, _m| None)
676 .mount("/v", Module::new("v").route("/", get(detailed)))
677 .build()
678 .unwrap();
679 let body = body_string(dispatch(&built, http::Method::GET, "/v/", "").await).await;
680 assert!(
681 body.contains("JC0422") && body.contains("details") && body.contains("\"field\""),
682 "None must keep the default body with details: {body}"
683 );
684 }
685
686 #[test]
687 fn conflicting_routes_fail_at_build_not_at_request_time() {
688 let app = App::new()
689 .route("/x", get(|| async { "a" }))
690 .route("/x", get(|| async { "b" }));
691 let err = app.build().unwrap_err();
692 assert!(err.message().contains("/x"));
693 }
694
695 #[test]
696 fn wildcard_origin_with_credentials_is_a_build_error() {
697 let err = App::new()
698 .cors(
699 crate::cors::CorsConfig::new(crate::cors::CorsOrigins::any())
700 .allow_credentials(true),
701 )
702 .build()
703 .unwrap_err();
704 assert!(
705 err.to_string().to_lowercase().contains("credential"),
706 "{err}"
707 );
708 }
709
710 #[test]
711 fn allowlist_origin_with_credentials_builds() {
712 assert!(
713 App::new()
714 .cors(
715 crate::cors::CorsConfig::new(crate::cors::CorsOrigins::list([
716 "https://app.example"
717 ]))
718 .allow_credentials(true)
719 )
720 .build()
721 .is_ok()
722 );
723 }
724
725 #[tokio::test]
726 async fn extensions_register_through_extend() {
727 struct Greeting(&'static str);
728 struct GreetingExt;
729 impl Extension for GreetingExt {
730 fn register(self, app: App) -> App {
731 app.provide(Greeting("from-extension"))
732 }
733 }
734 async fn read(g: crate::Dep<Greeting>) -> String {
735 (*g).0.to_string()
738 }
739 let t = App::new()
740 .extend(GreetingExt)
741 .route("/", crate::router::get(read))
742 .into_test();
743 assert_eq!(t.get("/").await.text(), "from-extension");
744 }
745
746 #[test]
747 fn accept_error_classification_matches_unix_reality() {
748 use std::io::{Error as IoError, ErrorKind};
749 for transient in [
750 IoError::from(ErrorKind::ConnectionAborted),
751 IoError::from(ErrorKind::ConnectionReset),
752 IoError::from(ErrorKind::Interrupted),
753 IoError::from_raw_os_error(24), IoError::from_raw_os_error(23), ] {
756 assert!(is_transient_accept_error(&transient), "{transient:?}");
757 }
758 assert!(!is_transient_accept_error(&IoError::from(
759 ErrorKind::InvalidInput
760 )));
761 assert!(!is_transient_accept_error(&IoError::from(
762 ErrorKind::PermissionDenied
763 )));
764 }
765}