1use std::{
2 collections::BTreeMap,
3 ffi::{OsStr, OsString},
4 sync::Arc,
5 time::Instant,
6};
7
8use actix_web::{
9 App, HttpRequest, HttpResponse, HttpServer,
10 body::{BodySize, MessageBody},
11 dev::{Service, ServiceResponse},
12 error::{JsonPayloadError, PathError, QueryPayloadError},
13 http::header,
14 web,
15};
16#[cfg(feature = "uploads")]
17use caelix_core::UploadConfig;
18#[cfg(feature = "openapi")]
19use caelix_core::openapi::{OpenApiConfig, build_openapi};
20use caelix_core::{
21 BadRequestException, BoxFuture, Container, HttpException, HttpResponse as CaelixHttpResponse,
22 IntoCaelixResponse, Module, NotFoundException, PayloadTooLargeException, ResponseBody, Result,
23 build_container, http_request_logging_enabled, log_application_starting, log_http_request,
24 log_http_request_info, log_module_routes, log_ready,
25 register_module_controllers_with_container, shutdown_module,
26};
27use futures_util::StreamExt;
28
29pub const DEFAULT_BODY_LIMIT_BYTES: usize = 1024 * 1024;
31
32#[derive(Clone)]
34pub(crate) struct UploadRuntimeConfig {
35 #[cfg(feature = "uploads")]
36 pub(crate) config: UploadConfig,
37 pub(crate) body_limit: usize,
38}
39
40#[cfg(feature = "openapi")]
41#[derive(Clone)]
42pub(crate) struct OpenApiServices {
43 pub config: OpenApiConfig,
45 pub document: String,
47}
48
49#[cfg(not(feature = "openapi"))]
50#[derive(Clone)]
51pub(crate) struct OpenApiServices;
52
53#[derive(Clone, Copy, Debug, Eq, PartialEq)]
54enum AccessLogFormat {
55 Compact,
56 Info,
57}
58
59#[derive(Clone, Copy, Debug, Eq, PartialEq)]
63pub struct Logging {
65 access_log: bool,
66 access_log_format: AccessLogFormat,
67}
68
69impl Default for Logging {
70 fn default() -> Self {
71 Self {
72 access_log: true,
73 access_log_format: AccessLogFormat::Compact,
74 }
75 }
76}
77
78impl Logging {
79 pub fn info() -> Self {
84 Self {
85 access_log: true,
86 access_log_format: AccessLogFormat::Info,
87 }
88 }
89
90 pub fn access_log(mut self, enabled: bool) -> Self {
92 self.access_log = enabled;
93 self
94 }
95
96 pub fn access_log_enabled(&self) -> bool {
98 self.access_log
99 }
100
101 fn access_log_format(&self) -> AccessLogFormat {
102 self.access_log_format
103 }
104}
105
106pub fn to_actix_response(response: CaelixHttpResponse) -> HttpResponse {
108 let status = actix_web::http::StatusCode::from_u16(response.status.as_u16())
110 .unwrap_or(actix_web::http::StatusCode::INTERNAL_SERVER_ERROR);
111
112 let mut builder = HttpResponse::build(status);
113 builder.content_type(response.content_type);
114 for (name, value) in response.headers {
115 builder.insert_header((name, value));
116 }
117 for cookie in response.cookies {
118 let mut runtime_cookie =
119 actix_web::cookie::Cookie::new(cookie.name().to_string(), cookie.value().to_string());
120 runtime_cookie.set_http_only(cookie.is_http_only());
121 runtime_cookie.set_secure(cookie.is_secure());
122 runtime_cookie.set_same_site(match cookie.same_site_value() {
123 caelix_core::SameSite::Strict => actix_web::cookie::SameSite::Strict,
124 caelix_core::SameSite::Lax => actix_web::cookie::SameSite::Lax,
125 caelix_core::SameSite::None => actix_web::cookie::SameSite::None,
126 });
127 if let Some(path) = cookie.path_value() {
128 runtime_cookie.set_path(path.to_string());
129 }
130 if let Some(domain) = cookie.domain_value() {
131 runtime_cookie.set_domain(domain.to_string());
132 }
133 if let Some(max_age) = cookie.max_age_value() {
134 runtime_cookie.set_max_age(
135 actix_web::cookie::time::Duration::try_from(max_age)
136 .unwrap_or(actix_web::cookie::time::Duration::MAX),
137 );
138 }
139 if let Some(expires) = cookie.expires_value() {
140 runtime_cookie.set_expires(actix_web::cookie::time::OffsetDateTime::from(expires));
141 }
142 builder.append_header((
143 actix_web::http::header::SET_COOKIE,
144 runtime_cookie.encoded().to_string(),
145 ));
146 }
147
148 match response.body {
149 ResponseBody::Buffered(bytes) => builder.body(bytes),
150 ResponseBody::Streaming(stream) => {
151 let stream = stream.map(|chunk| {
153 chunk.map_err(|err| {
154 caelix_core::log_http_exception(&err);
155 actix_web::error::ErrorInternalServerError("Internal Server Error")
156 })
157 });
158 builder.streaming(stream)
159 }
160 }
161}
162
163pub struct Application {
165 startup_started: Instant,
166 container: Arc<Container>,
167 configure_fn: fn(&mut web::ServiceConfig, Arc<Container>),
168 gateway_configure_fn: fn(&mut web::ServiceConfig, Arc<Container>, usize),
169 shutdown_fn: for<'a> fn(&'a Container) -> BoxFuture<'a, caelix_core::Result<()>>,
170 body_limit: usize,
171 #[cfg(feature = "uploads")]
172 upload_config: UploadConfig,
173 websocket_max_message_size: usize,
174 workers: usize,
175 logging: Option<Logging>,
176 openapi: Option<OpenApiServices>,
177 #[cfg(feature = "openapi")]
178 openapi_build_fn:
179 fn(&OpenApiConfig) -> caelix_core::Result<caelix_core::openapi::utoipa::openapi::OpenApi>,
180}
181
182fn json_config(body_limit: usize) -> web::JsonConfig {
183 web::JsonConfig::default()
184 .limit(body_limit)
185 .content_type_required(false)
186 .error_handler(move |err, _req| {
187 let exception = json_exception(&err, body_limit);
188 let response = to_actix_response(exception.into_response());
189
190 actix_web::error::InternalError::from_response(err, response).into()
191 })
192}
193
194fn json_exception(err: &JsonPayloadError, body_limit: usize) -> HttpException {
195 if matches!(
196 err,
197 JsonPayloadError::Overflow { .. } | JsonPayloadError::OverflowKnownLength { .. }
198 ) {
199 return PayloadTooLargeException::new(format!(
200 "request body exceeds the configured limit of {body_limit} bytes"
201 ));
202 }
203
204 if let JsonPayloadError::Deserialize(source) = err {
205 if let Some(exception) = missing_field_exception(&source.to_string()) {
206 return exception;
207 }
208 }
209
210 BadRequestException::new("invalid JSON request body")
211}
212
213fn path_config() -> web::PathConfig {
214 web::PathConfig::default().error_handler(|err: PathError, _req| {
215 let exception = missing_field_exception(&err.to_string())
216 .unwrap_or_else(|| BadRequestException::new(err.to_string()));
217 let response = to_actix_response(exception.into_response());
218
219 actix_web::error::InternalError::from_response(err, response).into()
220 })
221}
222
223fn query_config() -> web::QueryConfig {
224 web::QueryConfig::default().error_handler(|err: QueryPayloadError, _req| {
225 let exception = missing_field_exception(&err.to_string())
226 .unwrap_or_else(|| BadRequestException::new(err.to_string()));
227 let response = to_actix_response(exception.into_response());
228
229 actix_web::error::InternalError::from_response(err, response).into()
230 })
231}
232
233fn missing_field_exception(message: &str) -> Option<HttpException> {
234 let field = missing_field_name(message)?;
235 let mut errors = BTreeMap::new();
236 errors.insert(field, vec!["is required".to_string()]);
237
238 Some(BadRequestException::new("Validation failed").with_errors(errors))
239}
240
241fn missing_field_name(message: &str) -> Option<String> {
242 let start = message.find("missing field `")? + "missing field `".len();
243 let rest = &message[start..];
244 let end = rest.find('`')?;
245 let field = &rest[..end];
246
247 if field.is_empty() {
248 None
249 } else {
250 Some(field.to_string())
251 }
252}
253
254async fn not_found(req: HttpRequest) -> HttpResponse {
255 to_actix_response(
256 NotFoundException::new(format!("Cannot {} {}", req.method(), req.path())).into_response(),
257 )
258}
259
260fn log_access_request<B: MessageBody>(
261 format: AccessLogFormat,
262 response: &ServiceResponse<B>,
263 elapsed: std::time::Duration,
264) {
265 let request = response.request();
266
267 match format {
268 AccessLogFormat::Compact => log_http_request(
269 request.method().as_str(),
270 request.path(),
271 response.status().as_u16(),
272 elapsed,
273 ),
274 AccessLogFormat::Info => {
275 let path_and_query = if request.query_string().is_empty() {
276 request.path().to_string()
277 } else {
278 format!("{}?{}", request.path(), request.query_string())
279 };
280 let response_size = match response.response().body().size() {
281 BodySize::None => Some(0),
282 BodySize::Sized(size) => Some(size),
283 BodySize::Stream => None,
284 };
285
286 log_http_request_info(
287 request.connection_info().peer_addr().unwrap_or("-"),
288 request.method().as_str(),
289 &path_and_query,
290 &format!("{:?}", request.version()),
291 response.status().as_u16(),
292 response_size,
293 request_header(request, &header::REFERER).as_str(),
294 request_header(request, &header::USER_AGENT).as_str(),
295 elapsed,
296 );
297 }
298 }
299}
300
301fn request_header(request: &HttpRequest, name: &header::HeaderName) -> String {
302 request
303 .headers()
304 .get(name)
305 .map(|value| String::from_utf8_lossy(value.as_bytes()).into_owned())
306 .unwrap_or_else(|| "-".to_string())
307}
308
309pub(crate) fn configure_caelix_services(
310 cfg: &mut web::ServiceConfig,
311 body_limit: usize,
312 #[cfg(feature = "uploads")] upload_config: UploadConfig,
313 configure_fn: fn(&mut web::ServiceConfig, Arc<Container>),
314 container: Arc<Container>,
315 openapi: Option<&OpenApiServices>,
316) {
317 cfg.app_data(json_config(body_limit));
318 cfg.app_data(web::Data::new(UploadRuntimeConfig {
319 #[cfg(feature = "uploads")]
320 config: upload_config,
321 body_limit,
322 }));
323 cfg.app_data(path_config());
324 cfg.app_data(query_config());
325 configure_fn(cfg, container);
326 #[cfg(feature = "openapi")]
327 if let Some(openapi) = openapi {
328 let ui_base = openapi.config.ui_path.trim_end_matches('/');
329 let ui_redirect = format!("{ui_base}/");
330 let document = openapi.document.clone();
331 cfg.route(
332 &openapi.config.json_path,
333 web::get().to(move || {
334 let document = document.clone();
335 async move {
336 HttpResponse::Ok()
337 .content_type("application/json")
338 .body(document)
339 }
340 }),
341 );
342 let html = swagger_ui_html(&openapi.config.json_path);
343 cfg.route(
344 &format!("{ui_base}/"),
345 web::get().to(move || {
346 let html = html.clone();
347 async move {
348 HttpResponse::Ok()
349 .content_type("text/html; charset=utf-8")
350 .body(html)
351 }
352 }),
353 );
354 cfg.route(
355 ui_base,
356 web::get().to(move || {
357 let ui_redirect = ui_redirect.clone();
358 async move {
359 HttpResponse::TemporaryRedirect()
360 .insert_header((header::LOCATION, ui_redirect))
361 .finish()
362 }
363 }),
364 );
365 }
366 #[cfg(not(feature = "openapi"))]
367 let _ = openapi;
368 cfg.default_service(web::route().to(not_found));
369}
370
371#[cfg(feature = "openapi")]
372fn swagger_ui_html(json_path: &str) -> String {
373 let json_path = serde_json::to_string(json_path).expect("OpenAPI path must serialize");
374 format!(
375 r#"<!doctype html><html><head><meta charset="utf-8"><title>Swagger UI</title><link rel="stylesheet" href="https://unpkg.com/swagger-ui-dist@5/swagger-ui.css"></head><body><div id="swagger-ui"></div><script src="https://unpkg.com/swagger-ui-dist@5/swagger-ui-bundle.js"></script><script>SwaggerUIBundle({{url:{json_path},dom_id:'#swagger-ui'}});</script></body></html>"#
376 )
377}
378
379impl Application {
380 pub async fn new<M: Module + 'static>() -> Result<Self> {
382 let start = Instant::now();
383 log_application_starting();
384 let container = build_container::<M>().await?;
385 log_module_routes::<M>();
386
387 Ok(Self {
388 startup_started: start,
389 container: Arc::new(container),
390 configure_fn: |cfg, container| {
391 register_module_controllers_with_container::<M>(cfg, container)
392 },
393 gateway_configure_fn: |cfg, container, max| {
394 crate::websocket::configure_gateway_routes::<M>(cfg, container, max)
395 },
396 shutdown_fn: |container| Box::pin(async move { shutdown_module::<M>(container).await }),
397 body_limit: DEFAULT_BODY_LIMIT_BYTES,
398 #[cfg(feature = "uploads")]
399 upload_config: UploadConfig::default(),
400 websocket_max_message_size: crate::websocket::DEFAULT_WEBSOCKET_MAX_MESSAGE_SIZE,
401 workers: num_cpus::get(),
402 logging: None,
403 openapi: None,
404 #[cfg(feature = "openapi")]
405 openapi_build_fn: |config| build_openapi::<M>(config),
406 })
407 }
408
409 pub fn body_limit(mut self, bytes: usize) -> Self {
411 self.body_limit = bytes;
412 self
413 }
414
415 #[cfg(feature = "uploads")]
416 pub fn upload_temp_dir(mut self, path: impl Into<std::path::PathBuf>) -> Self {
418 self.upload_config = self.upload_config.upload_temp_dir(path);
419 self
420 }
421
422 pub fn websocket_max_message_size(mut self, bytes: usize) -> Self {
424 self.websocket_max_message_size = bytes.max(1);
425 self
426 }
427
428 pub fn workers(mut self, workers: usize) -> Self {
430 self.workers = workers.max(1);
431 self
432 }
433
434 pub fn logging(mut self, logging: Logging) -> Self {
440 self.logging = Some(logging);
441 self
442 }
443
444 #[cfg(feature = "openapi")]
446 pub fn with_openapi(mut self, config: OpenApiConfig) -> Result<Self> {
448 let document = (self.openapi_build_fn)(&config)?;
449 self.openapi = Some(OpenApiServices {
450 config,
451 document: document.to_json().expect("OpenAPI document must serialize"),
452 });
453 Ok(self)
454 }
455
456 #[cfg(test)]
457 fn configure_services(&self, cfg: &mut web::ServiceConfig) {
458 configure_caelix_services(
459 cfg,
460 self.body_limit,
461 #[cfg(feature = "uploads")]
462 self.upload_config.clone(),
463 self.configure_fn,
464 self.container.clone(),
465 self.openapi.as_ref(),
466 );
467 }
468
469 async fn shutdown(&self) -> caelix_core::Result<()> {
470 (self.shutdown_fn)(&self.container).await
471 }
472
473 fn prepare_doctor_runtime(&self) {
474 let container = self.container.clone();
475 let configure_fn = self.configure_fn;
476 let body_limit = self.body_limit;
477 #[cfg(feature = "uploads")]
478 let upload_config = self.upload_config.clone();
479 let websocket_max_message_size = self.websocket_max_message_size;
480 let gateway_configure_fn = self.gateway_configure_fn;
481 let openapi = self.openapi.clone();
482 let route_container = container.clone();
483
484 let _app = App::new()
485 .app_data(web::Data::from(container.clone()))
486 .configure({
487 move |cfg| {
488 configure_caelix_services(
489 cfg,
490 body_limit,
491 #[cfg(feature = "uploads")]
492 upload_config.clone(),
493 configure_fn,
494 route_container.clone(),
495 openapi.as_ref(),
496 )
497 }
498 })
499 .configure(move |cfg| {
500 gateway_configure_fn(cfg, container.clone(), websocket_max_message_size)
501 });
502 }
503
504 pub async fn listen(self, addr: &str) -> std::io::Result<()> {
506 self.listen_with_doctor_mode(addr, has_doctor_argument(std::env::args_os()))
507 .await
508 }
509
510 async fn listen_with_doctor_mode(self, addr: &str, doctor_mode: bool) -> std::io::Result<()> {
511 if doctor_mode {
512 self.prepare_doctor_runtime();
513 return self.shutdown().await.map_err(to_io_error);
514 }
515
516 let container = self.container.clone();
517 let configure_fn = self.configure_fn;
518 let body_limit = self.body_limit;
519 #[cfg(feature = "uploads")]
520 let upload_config = self.upload_config.clone();
521 let websocket_max_message_size = self.websocket_max_message_size;
522 let gateway_configure_fn = self.gateway_configure_fn;
523 let workers = self.workers;
524 let addr = addr.to_string();
525 let logging = self.logging.unwrap_or(Logging {
526 access_log: http_request_logging_enabled(),
527 access_log_format: AccessLogFormat::Compact,
528 });
529 let openapi = self.openapi.clone();
530
531 let result = if logging.access_log_enabled() {
532 let logging_container = container.clone();
533 let openapi_with_logging = openapi.clone();
534 let access_log_format = logging.access_log_format();
535 let server = match HttpServer::new(move || {
536 let route_container = logging_container.clone();
537 App::new()
538 .app_data(web::Data::from(logging_container.clone()))
539 .wrap_fn(move |req, service| {
540 let request_log_start = Instant::now();
541 let future = service.call(req);
542
543 async move {
544 let response = future.await?;
545 log_access_request(
546 access_log_format,
547 &response,
548 request_log_start.elapsed(),
549 );
550 Ok(response)
551 }
552 })
553 .configure({
554 let openapi = openapi_with_logging.clone();
555 #[cfg(feature = "uploads")]
556 let upload_config = upload_config.clone();
557 move |cfg| {
558 configure_caelix_services(
559 cfg,
560 body_limit,
561 #[cfg(feature = "uploads")]
562 upload_config.clone(),
563 configure_fn,
564 route_container.clone(),
565 openapi.as_ref(),
566 )
567 }
568 })
569 .configure({
570 let container = logging_container.clone();
571 move |cfg| {
572 gateway_configure_fn(cfg, container.clone(), websocket_max_message_size)
573 }
574 })
575 })
576 .workers(workers)
577 .bind(addr.as_str())
578 {
579 Ok(server) => {
580 log_ready(&addr, self.startup_started.elapsed());
581 server.run()
582 }
583 Err(err) => {
584 let _ = self.shutdown().await;
585 return Err(err);
586 }
587 };
588
589 server.await
590 } else {
591 let server = match HttpServer::new(move || {
592 let route_container = container.clone();
593 App::new()
594 .app_data(web::Data::from(container.clone()))
595 .configure({
596 let openapi = openapi.clone();
597 #[cfg(feature = "uploads")]
598 let upload_config = upload_config.clone();
599 move |cfg| {
600 configure_caelix_services(
601 cfg,
602 body_limit,
603 #[cfg(feature = "uploads")]
604 upload_config.clone(),
605 configure_fn,
606 route_container.clone(),
607 openapi.as_ref(),
608 )
609 }
610 })
611 .configure({
612 let container = container.clone();
613 move |cfg| {
614 gateway_configure_fn(cfg, container.clone(), websocket_max_message_size)
615 }
616 })
617 })
618 .workers(workers)
619 .bind(addr.as_str())
620 {
621 Ok(server) => {
622 log_ready(&addr, self.startup_started.elapsed());
623 server.run()
624 }
625 Err(err) => {
626 let _ = self.shutdown().await;
627 return Err(err);
628 }
629 };
630
631 server.await
632 };
633
634 self.shutdown().await.map_err(to_io_error)?;
635 result
636 }
637}
638
639fn has_doctor_argument<I>(args: I) -> bool
640where
641 I: IntoIterator<Item = OsString>,
642{
643 args.into_iter().any(|arg| arg == OsStr::new("--doctor"))
644}
645
646fn to_io_error(err: caelix_core::HttpException) -> std::io::Error {
647 std::io::Error::other(err.message)
648}
649
650#[cfg(test)]
651mod tests {
652 use super::*;
653 use actix_web::{http::StatusCode, test as actix_test};
654 use caelix_core::{Controller, Injectable, ModuleMetadata};
655 use serde::Deserialize;
656 use serde_json::{Value, json};
657 use std::{
658 any::Any,
659 sync::atomic::{AtomicUsize, Ordering},
660 };
661
662 #[test]
663 fn response_adapter_appends_every_cookie_header() {
664 let response = to_actix_response(
665 CaelixHttpResponse::text(caelix_core::StatusCode::OK, "ok")
666 .with_cookie(caelix_core::Cookie::new("session", "a b"))
667 .with_cookie(
668 caelix_core::Cookie::removal("preference")
669 .path("/settings")
670 .domain("example.com"),
671 ),
672 );
673 let values = response
674 .headers()
675 .get_all(actix_web::http::header::SET_COOKIE)
676 .into_iter()
677 .map(|value| value.to_str().unwrap().to_string())
678 .collect::<Vec<_>>();
679 assert_eq!(values.len(), 2);
680 assert!(values[0].contains("session=a%20b"));
681 assert!(values[0].contains("HttpOnly"));
682 assert!(values[0].contains("Secure"));
683 assert!(values[0].contains("SameSite=Lax"));
684 assert!(values[1].contains("Max-Age=0"));
685 assert!(values[1].contains("Domain=example.com"));
686 assert!(values[1].contains("Path=/settings"));
687 }
688 use uuid::Uuid;
689
690 static SHUTDOWN_COUNT: AtomicUsize = AtomicUsize::new(0);
691 static DOCTOR_CONSTRUCTION_COUNT: AtomicUsize = AtomicUsize::new(0);
692 static DOCTOR_INIT_COUNT: AtomicUsize = AtomicUsize::new(0);
693 static DOCTOR_STARTUP_COUNT: AtomicUsize = AtomicUsize::new(0);
694 static DOCTOR_SHUTDOWN_COUNT: AtomicUsize = AtomicUsize::new(0);
695 static DOCTOR_ROUTE_CONFIG_COUNT: AtomicUsize = AtomicUsize::new(0);
696
697 struct HealthService {
698 status: &'static str,
699 }
700
701 impl Injectable for HealthService {
702 fn dependencies() -> Vec<caelix_core::ProviderDependency> {
703 caelix_core::provider_dependencies![]
704 }
705
706 fn create(_container: &Container) -> caelix_core::BoxFuture<'_, caelix_core::Result<Self>> {
707 Box::pin(async move { Ok(Self { status: "ok" }) })
708 }
709 }
710
711 struct TestModule;
712
713 impl Module for TestModule {
714 fn register() -> ModuleMetadata {
715 ModuleMetadata::new().provider::<HealthService>()
716 }
717 }
718
719 struct JsonController;
720
721 impl Injectable for JsonController {
722 fn dependencies() -> Vec<caelix_core::ProviderDependency> {
723 caelix_core::provider_dependencies![]
724 }
725
726 fn create(_container: &Container) -> caelix_core::BoxFuture<'_, caelix_core::Result<Self>> {
727 Box::pin(async move { Ok(Self) })
728 }
729 }
730
731 impl JsonController {
732 async fn accept_json(_payload: web::Json<Value>) -> HttpResponse {
733 HttpResponse::Ok().finish()
734 }
735 }
736
737 impl Controller for JsonController {
738 fn base_path() -> &'static str {
739 "/json"
740 }
741
742 fn register_routes(cfg_any: &mut dyn Any) {
743 let cfg = cfg_any
744 .downcast_mut::<web::ServiceConfig>()
745 .expect("expected actix ServiceConfig");
746
747 cfg.route("/json", web::post().to(Self::accept_json));
748 }
749 }
750
751 struct JsonModule;
752
753 impl Module for JsonModule {
754 fn register() -> ModuleMetadata {
755 ModuleMetadata::new().controller::<JsonController>()
756 }
757 }
758
759 #[derive(Deserialize)]
760 struct SearchQuery {
761 limit: usize,
762 }
763
764 #[derive(Deserialize)]
765 struct RequiredBody {
766 name: String,
767 }
768
769 #[derive(Deserialize)]
770 struct RequiredQuery {
771 q: String,
772 }
773
774 #[derive(Deserialize)]
775 struct RequiredPath {
776 org_id: Uuid,
777 user_id: Uuid,
778 }
779
780 async fn accept_uuid(_id: web::Path<Uuid>) -> HttpResponse {
781 HttpResponse::Ok().finish()
782 }
783
784 async fn accept_required_body(body: web::Json<RequiredBody>) -> HttpResponse {
785 let body = body.into_inner();
786 let _ = body.name;
787
788 HttpResponse::Ok().finish()
789 }
790
791 async fn accept_query(query: web::Query<SearchQuery>) -> HttpResponse {
792 let query = query.into_inner();
793 let _ = query.limit;
794
795 HttpResponse::Ok().finish()
796 }
797
798 async fn accept_required_query(query: web::Query<RequiredQuery>) -> HttpResponse {
799 let query = query.into_inner();
800 let _ = query.q;
801
802 HttpResponse::Ok().finish()
803 }
804
805 async fn accept_required_path(path: web::Path<RequiredPath>) -> HttpResponse {
806 let path = path.into_inner();
807 let _ = (path.org_id, path.user_id);
808
809 HttpResponse::Ok().finish()
810 }
811
812 struct ShutdownService;
813
814 impl Injectable for ShutdownService {
815 fn dependencies() -> Vec<caelix_core::ProviderDependency> {
816 caelix_core::provider_dependencies![]
817 }
818
819 fn create(_container: &Container) -> caelix_core::BoxFuture<'_, caelix_core::Result<Self>> {
820 Box::pin(async move { Ok(Self) })
821 }
822
823 fn on_shutdown(&self) -> caelix_core::BoxFuture<'_, caelix_core::Result<()>> {
824 Box::pin(async move {
825 SHUTDOWN_COUNT.fetch_add(1, Ordering::SeqCst);
826 Ok(())
827 })
828 }
829 }
830
831 struct ShutdownModule;
832
833 impl Module for ShutdownModule {
834 fn register() -> ModuleMetadata {
835 ModuleMetadata::new().provider::<ShutdownService>()
836 }
837 }
838
839 struct DoctorService;
840
841 impl Injectable for DoctorService {
842 fn dependencies() -> Vec<caelix_core::ProviderDependency> {
843 caelix_core::provider_dependencies![]
844 }
845
846 fn create(_container: &Container) -> caelix_core::BoxFuture<'_, caelix_core::Result<Self>> {
847 Box::pin(async move { Ok(Self) })
848 }
849
850 fn on_bootstrap(&self) -> caelix_core::BoxFuture<'_, caelix_core::Result<()>> {
851 Box::pin(async move {
852 DOCTOR_STARTUP_COUNT.fetch_add(1, Ordering::SeqCst);
853 Ok(())
854 })
855 }
856
857 fn on_shutdown(&self) -> caelix_core::BoxFuture<'_, caelix_core::Result<()>> {
858 Box::pin(async move {
859 DOCTOR_SHUTDOWN_COUNT.fetch_add(1, Ordering::SeqCst);
860 Ok(())
861 })
862 }
863 }
864
865 struct DoctorController;
866
867 impl Injectable for DoctorController {
868 fn dependencies() -> Vec<caelix_core::ProviderDependency> {
869 caelix_core::provider_dependencies![]
870 }
871
872 fn create(_container: &Container) -> caelix_core::BoxFuture<'_, caelix_core::Result<Self>> {
873 Box::pin(async move {
874 DOCTOR_CONSTRUCTION_COUNT.fetch_add(1, Ordering::SeqCst);
875 Ok(Self)
876 })
877 }
878
879 fn on_module_init(&self) -> caelix_core::BoxFuture<'_, caelix_core::Result<()>> {
880 Box::pin(async move {
881 DOCTOR_INIT_COUNT.fetch_add(1, Ordering::SeqCst);
882 Ok(())
883 })
884 }
885 }
886
887 impl Controller for DoctorController {
888 fn base_path() -> &'static str {
889 "/doctor"
890 }
891
892 fn register_routes(cfg_any: &mut dyn Any) {
893 DOCTOR_ROUTE_CONFIG_COUNT.fetch_add(1, Ordering::SeqCst);
894 let cfg = cfg_any
895 .downcast_mut::<web::ServiceConfig>()
896 .expect("expected actix ServiceConfig");
897 cfg.route(
898 "/doctor",
899 web::get().to(|| async { HttpResponse::Ok().finish() }),
900 );
901 }
902 }
903
904 struct DoctorModule;
905
906 impl Module for DoctorModule {
907 fn register() -> ModuleMetadata {
908 ModuleMetadata::new()
909 .provider::<DoctorService>()
910 .controller::<DoctorController>()
911 }
912 }
913
914 struct FailingShutdownService;
915
916 impl Injectable for FailingShutdownService {
917 fn dependencies() -> Vec<caelix_core::ProviderDependency> {
918 caelix_core::provider_dependencies![]
919 }
920
921 fn create(_container: &Container) -> caelix_core::BoxFuture<'_, caelix_core::Result<Self>> {
922 Box::pin(async move { Ok(Self) })
923 }
924
925 fn on_shutdown(&self) -> caelix_core::BoxFuture<'_, caelix_core::Result<()>> {
926 Box::pin(async move {
927 Err(caelix_core::HttpException::new(
928 caelix_core::StatusCode::INTERNAL_SERVER_ERROR,
929 "Internal Server Error",
930 "shutdown failed",
931 ))
932 })
933 }
934 }
935
936 struct FailingShutdownModule;
937
938 impl Module for FailingShutdownModule {
939 fn register() -> ModuleMetadata {
940 ModuleMetadata::new().provider::<FailingShutdownService>()
941 }
942 }
943
944 #[actix_web::test]
945 async fn new_builds_container_from_module_metadata() {
946 let app = Application::new::<TestModule>().await.unwrap();
947
948 let service = app.container.resolve::<HealthService>().unwrap();
949
950 assert_eq!(service.status, "ok");
951 }
952
953 #[actix_web::test]
954 async fn application_accepts_explicit_logging_configuration() {
955 let app = Application::new::<TestModule>()
956 .await
957 .unwrap()
958 .logging(Logging::default().access_log(false));
959
960 assert_eq!(app.logging, Some(Logging::default().access_log(false)));
961 assert!(!Logging::default().access_log(false).access_log_enabled());
962 assert!(Logging::default().access_log_enabled());
963 assert_eq!(Logging::info().access_log_format(), AccessLogFormat::Info);
964 }
965
966 #[actix_web::test]
967 async fn json_body_limit_rejects_large_payloads_with_json_error() {
968 async fn accept_json(_payload: web::Json<Value>) -> HttpResponse {
969 HttpResponse::Ok().finish()
970 }
971
972 let app = actix_test::init_service(
973 App::new()
974 .app_data(json_config(8))
975 .route("/json", web::post().to(accept_json)),
976 )
977 .await;
978
979 let response = actix_test::call_service(
980 &app,
981 actix_test::TestRequest::post()
982 .uri("/json")
983 .insert_header(("content-type", "application/json"))
984 .set_payload(r#"{"too":"large"}"#)
985 .to_request(),
986 )
987 .await;
988
989 assert_eq!(response.status(), StatusCode::PAYLOAD_TOO_LARGE);
990 let body: Value = actix_test::read_body_json(response).await;
991 assert_eq!(
992 body,
993 json!({
994 "status": 413,
995 "error": "Payload Too Large",
996 "message": "request body exceeds the configured limit of 8 bytes"
997 })
998 );
999 }
1000
1001 #[actix_web::test]
1002 async fn json_config_accepts_json_without_content_type_header() {
1003 async fn accept_json(_payload: web::Json<Value>) -> HttpResponse {
1004 HttpResponse::Ok().finish()
1005 }
1006
1007 let app = actix_test::init_service(
1008 App::new()
1009 .app_data(json_config(DEFAULT_BODY_LIMIT_BYTES))
1010 .route("/json", web::post().to(accept_json)),
1011 )
1012 .await;
1013
1014 let response = actix_test::call_service(
1015 &app,
1016 actix_test::TestRequest::post()
1017 .uri("/json")
1018 .set_payload("{}")
1019 .to_request(),
1020 )
1021 .await;
1022
1023 assert_eq!(response.status(), StatusCode::OK);
1024 }
1025
1026 #[actix_web::test]
1027 async fn json_missing_field_errors_are_validation_shaped() {
1028 let app = actix_test::init_service(
1029 App::new()
1030 .app_data(json_config(DEFAULT_BODY_LIMIT_BYTES))
1031 .route("/json", web::patch().to(accept_required_body)),
1032 )
1033 .await;
1034
1035 let response = actix_test::call_service(
1036 &app,
1037 actix_test::TestRequest::patch()
1038 .uri("/json")
1039 .insert_header(("content-type", "application/json"))
1040 .set_payload("{}")
1041 .to_request(),
1042 )
1043 .await;
1044
1045 assert_eq!(response.status(), StatusCode::BAD_REQUEST);
1046 let body: Value = actix_test::read_body_json(response).await;
1047 assert_eq!(
1048 body,
1049 json!({
1050 "status": 400,
1051 "error": "Bad Request",
1052 "message": "Validation failed",
1053 "errors": {
1054 "name": ["is required"]
1055 }
1056 })
1057 );
1058 }
1059
1060 #[actix_web::test]
1061 async fn application_enforces_configured_body_limit() {
1062 let application = Application::new::<JsonModule>()
1063 .await
1064 .unwrap()
1065 .body_limit(8);
1066 let app = actix_test::init_service(
1067 App::new()
1068 .app_data(web::Data::from(application.container.clone()))
1069 .configure(|cfg| application.configure_services(cfg)),
1070 )
1071 .await;
1072
1073 let response = actix_test::call_service(
1074 &app,
1075 actix_test::TestRequest::post()
1076 .uri("/json")
1077 .insert_header(("content-type", "application/json"))
1078 .set_payload(r#"{"too":"large"}"#)
1079 .to_request(),
1080 )
1081 .await;
1082
1083 assert_eq!(response.status(), StatusCode::PAYLOAD_TOO_LARGE);
1084 let body: Value = actix_test::read_body_json(response).await;
1085 assert_eq!(
1086 body,
1087 json!({
1088 "status": 413,
1089 "error": "Payload Too Large",
1090 "message": "request body exceeds the configured limit of 8 bytes"
1091 })
1092 );
1093 }
1094
1095 #[actix_web::test]
1096 async fn path_extractor_errors_are_caelix_json_errors() {
1097 let app = actix_test::init_service(
1098 App::new()
1099 .app_data(path_config())
1100 .route("/users/{id}", web::get().to(accept_uuid)),
1101 )
1102 .await;
1103
1104 let response = actix_test::call_service(
1105 &app,
1106 actix_test::TestRequest::get().uri("/users/1").to_request(),
1107 )
1108 .await;
1109
1110 assert_eq!(response.status(), StatusCode::BAD_REQUEST);
1111 let body: Value = actix_test::read_body_json(response).await;
1112 assert_eq!(body["status"], 400);
1113 assert_eq!(body["error"], "Bad Request");
1114 assert!(
1115 body["message"]
1116 .as_str()
1117 .is_some_and(|message| message.contains("UUID parsing failed"))
1118 );
1119 }
1120
1121 #[actix_web::test]
1122 async fn path_missing_field_errors_are_validation_shaped() {
1123 let app = actix_test::init_service(App::new().app_data(path_config()).route(
1124 "/orgs/{org_id}/users/{user}",
1125 web::get().to(accept_required_path),
1126 ))
1127 .await;
1128
1129 let response = actix_test::call_service(
1130 &app,
1131 actix_test::TestRequest::get()
1132 .uri("/orgs/550e8400-e29b-41d4-a716-446655440000/users/550e8400-e29b-41d4-a716-446655440000")
1133 .to_request(),
1134 )
1135 .await;
1136
1137 assert_eq!(response.status(), StatusCode::BAD_REQUEST);
1138 let body: Value = actix_test::read_body_json(response).await;
1139 assert_eq!(
1140 body,
1141 json!({
1142 "status": 400,
1143 "error": "Bad Request",
1144 "message": "Validation failed",
1145 "errors": {
1146 "user_id": ["is required"]
1147 }
1148 })
1149 );
1150 }
1151
1152 #[actix_web::test]
1153 async fn query_extractor_errors_are_caelix_json_errors() {
1154 let app = actix_test::init_service(
1155 App::new()
1156 .app_data(query_config())
1157 .route("/users", web::get().to(accept_query)),
1158 )
1159 .await;
1160
1161 let response = actix_test::call_service(
1162 &app,
1163 actix_test::TestRequest::get()
1164 .uri("/users?limit=abc")
1165 .to_request(),
1166 )
1167 .await;
1168
1169 assert_eq!(response.status(), StatusCode::BAD_REQUEST);
1170 let body: Value = actix_test::read_body_json(response).await;
1171 assert_eq!(body["status"], 400);
1172 assert_eq!(body["error"], "Bad Request");
1173 assert!(
1174 body["message"]
1175 .as_str()
1176 .is_some_and(|message| message.contains("invalid digit"))
1177 );
1178 }
1179
1180 #[actix_web::test]
1181 async fn query_missing_field_errors_are_validation_shaped() {
1182 let app = actix_test::init_service(
1183 App::new()
1184 .app_data(query_config())
1185 .route("/users", web::get().to(accept_required_query)),
1186 )
1187 .await;
1188
1189 let response = actix_test::call_service(
1190 &app,
1191 actix_test::TestRequest::get().uri("/users").to_request(),
1192 )
1193 .await;
1194
1195 assert_eq!(response.status(), StatusCode::BAD_REQUEST);
1196 let body: Value = actix_test::read_body_json(response).await;
1197 assert_eq!(
1198 body,
1199 json!({
1200 "status": 400,
1201 "error": "Bad Request",
1202 "message": "Validation failed",
1203 "errors": {
1204 "q": ["is required"]
1205 }
1206 })
1207 );
1208 }
1209
1210 #[actix_web::test]
1211 async fn unmatched_routes_are_caelix_json_errors() {
1212 let app = actix_test::init_service(App::new().configure(|cfg| {
1213 configure_caelix_services(
1214 cfg,
1215 DEFAULT_BODY_LIMIT_BYTES,
1216 #[cfg(feature = "uploads")]
1217 UploadConfig::default(),
1218 |_, _| {},
1219 Arc::new(Container::new()),
1220 None,
1221 )
1222 }))
1223 .await;
1224
1225 let response = actix_test::call_service(
1226 &app,
1227 actix_test::TestRequest::get().uri("/missing").to_request(),
1228 )
1229 .await;
1230
1231 assert_eq!(response.status(), StatusCode::NOT_FOUND);
1232 let body: Value = actix_test::read_body_json(response).await;
1233 assert_eq!(
1234 body,
1235 json!({
1236 "status": 404,
1237 "error": "Not Found",
1238 "message": "Cannot GET /missing"
1239 })
1240 );
1241 }
1242
1243 #[actix_web::test]
1244 async fn application_runs_module_shutdown_hook() {
1245 SHUTDOWN_COUNT.store(0, Ordering::SeqCst);
1246
1247 let application = Application::new::<ShutdownModule>().await.unwrap();
1248 application.shutdown().await.unwrap();
1249
1250 assert_eq!(SHUTDOWN_COUNT.load(Ordering::SeqCst), 1);
1251 }
1252
1253 #[actix_web::test]
1254 async fn doctor_mode_runs_startup_runtime_setup_and_shutdown_without_binding() {
1255 DOCTOR_CONSTRUCTION_COUNT.store(0, Ordering::SeqCst);
1256 DOCTOR_INIT_COUNT.store(0, Ordering::SeqCst);
1257 DOCTOR_STARTUP_COUNT.store(0, Ordering::SeqCst);
1258 DOCTOR_SHUTDOWN_COUNT.store(0, Ordering::SeqCst);
1259 DOCTOR_ROUTE_CONFIG_COUNT.store(0, Ordering::SeqCst);
1260
1261 let application = Application::new::<DoctorModule>().await.unwrap();
1262 assert_eq!(DOCTOR_CONSTRUCTION_COUNT.load(Ordering::SeqCst), 1);
1263 assert_eq!(DOCTOR_INIT_COUNT.load(Ordering::SeqCst), 1);
1264 assert_eq!(DOCTOR_STARTUP_COUNT.load(Ordering::SeqCst), 1);
1265
1266 application
1267 .listen_with_doctor_mode("not a socket address", true)
1268 .await
1269 .unwrap();
1270
1271 assert_eq!(DOCTOR_ROUTE_CONFIG_COUNT.load(Ordering::SeqCst), 1);
1272 assert_eq!(DOCTOR_SHUTDOWN_COUNT.load(Ordering::SeqCst), 1);
1273 }
1274
1275 #[actix_web::test]
1276 async fn doctor_mode_propagates_shutdown_failures() {
1277 let error = Application::new::<FailingShutdownModule>()
1278 .await
1279 .unwrap()
1280 .listen_with_doctor_mode("not a socket address", true)
1281 .await
1282 .unwrap_err();
1283
1284 assert!(error.to_string().contains("shutdown failed"));
1285 }
1286
1287 #[actix_web::test]
1288 async fn normal_listen_still_attempts_to_bind_the_configured_address() {
1289 let error = Application::new::<TestModule>()
1290 .await
1291 .unwrap()
1292 .listen_with_doctor_mode("127.0.0.1:not-a-port", false)
1293 .await
1294 .unwrap_err();
1295
1296 assert!(error.to_string().contains("invalid port value"));
1297 }
1298
1299 #[test]
1300 fn doctor_mode_requires_the_exact_process_argument() {
1301 assert!(has_doctor_argument([OsString::from("--doctor")]));
1302 assert!(!has_doctor_argument([OsString::from("--doctor=true")]));
1303 assert!(!has_doctor_argument([OsString::from("doctor")]));
1304 }
1305
1306 #[actix_web::test]
1307 async fn to_actix_response_streams_chunked_body() {
1308 use actix_web::body::to_bytes;
1309 use caelix_core::{Bytes, Response};
1310
1311 let stream = futures_util::stream::iter(vec![
1312 Ok::<_, caelix_core::HttpException>(Bytes::from_static(b"chunk-a-")),
1313 Ok(Bytes::from_static(b"chunk-b")),
1314 ]);
1315 let caelix = Response::stream("text/plain", stream);
1316 let actix_response = to_actix_response(caelix);
1317
1318 assert_eq!(actix_response.status(), StatusCode::OK);
1319 assert_eq!(
1320 actix_response
1321 .headers()
1322 .get(actix_web::http::header::CONTENT_TYPE)
1323 .unwrap(),
1324 "text/plain"
1325 );
1326
1327 let body = to_bytes(actix_response.into_body()).await.unwrap();
1328 assert_eq!(&body[..], b"chunk-a-chunk-b");
1329 }
1330
1331 #[actix_web::test]
1332 async fn to_actix_response_applies_sse_headers() {
1333 use caelix_core::Response;
1334
1335 let stream = futures_util::stream::iter(Vec::<
1336 std::result::Result<serde_json::Value, caelix_core::HttpException>,
1337 >::new());
1338 let actix_response = to_actix_response(Response::sse(stream));
1339
1340 assert_eq!(
1341 actix_response
1342 .headers()
1343 .get(actix_web::http::header::CONTENT_TYPE)
1344 .unwrap(),
1345 "text/event-stream"
1346 );
1347 assert_eq!(
1348 actix_response.headers().get("cache-control").unwrap(),
1349 "no-cache"
1350 );
1351 assert_eq!(
1352 actix_response.headers().get("x-accel-buffering").unwrap(),
1353 "no"
1354 );
1355 }
1356}