1use hyper::body::Incoming;
8use hyper::service::service_fn;
9use hyper::Request;
10use hyper_util::rt::TokioIo;
11use rust_dix::{ServiceCollection, ServiceProvider};
12use webx_core::app::IHost;
13use webx_core::config::{self, AppOptions};
14use webx_core::error::Result;
15use webx_core::handler::IHostedService;
16use webx_core::http::IHttpContext;
17use webx_core::localization::{IStringLocalizer, LocalizationOptions, StringLocalizer};
18use webx_core::middleware::IMiddleware;
19use webx_core::mode::AppMode;
20use webx_core::route::catalog::{EndpointCatalog, IEndpointCatalog};
21use webx_core::routing::{HttpMethod, IEndpoint, IRouter};
22
23use crate::auth_jwt::{init_jwt_secret, jwt_middleware, JwtAuth};
24use crate::authz::collect_authorizers;
25use crate::context::{BodyLimits, HttpContext, RespBody};
26use crate::cors::{CorsConfig, CorsMiddleware};
27use crate::endpoint::{StaticHtmlEndpoint, StaticJsonEndpoint, StubEndpoint};
28use crate::health::{HealthCheckEndpoint, HealthCheckRegistry, HealthStatus};
29use crate::memory_cache::MemoryCache;
30use crate::metrics::{HttpMetrics, MetricsEndpoint, MetricsMiddleware};
31use crate::pipeline::{HandlerFn, MiddlewarePipeline};
32use crate::problem_response::{build_problem, write_problem_response};
33use crate::rate_limit::RateLimitMiddleware;
34use crate::router::Router;
35use jsonwebtoken::{DecodingKey, Validation};
36use webx_core::route::scan::RouteEntry;
37use webx_core::DispatchRuntime;
38use webx_openapi::{generate_openapi_spec, APIUI_HTML};
39use webx_spa::{EmbeddedAssets, SpaMiddleware, SpaSource};
40
41use rustls::pki_types::{CertificateDer, PrivateKeyDer};
42use rustls::ServerConfig;
43use rustls_pemfile::{certs, pkcs8_private_keys, rsa_private_keys};
44use std::sync::Arc;
45use std::time::Instant;
46use tokio::task::JoinSet;
47use tokio_rustls::TlsAcceptor;
48
49fn is_weak_jwt_secret(secret: &str) -> bool {
51 secret.is_empty()
52 || secret.len() < 32
53 || secret.contains("change-in-production")
54 || secret.contains("please-change")
55 || secret.contains("demo-secret")
56 || secret.contains("dev-secret")
57}
58
59fn log_spa_source(spa: &SpaMiddleware) {
64 tracing::info!("[Host] SPA root: {}", spa.resolved_root().display());
65 let Some(assets) = spa.embedded() else {
66 return;
67 };
68 tracing::info!(
69 "[Host] Embedded assets: {} files, {:.1} MiB embedded ({:.1} MiB uncompressed), built from `{}`",
70 assets.len(),
71 assets.total_bytes() as f64 / (1024.0 * 1024.0),
72 assets.total_raw_bytes() as f64 / (1024.0 * 1024.0),
73 assets.root(),
74 );
75 let shadowed = assets.shadowed_by(spa.resolved_root());
76 if !shadowed.is_empty() {
77 tracing::warn!(
78 "[Host] {} embedded file(s) are overridden on disk: {}. \
79 An override older than the current build stays in effect until it is \
80 updated or deleted.",
81 shadowed.len(),
82 shadowed.join(", "),
83 );
84 }
85}
86
87fn with_embedded_assets(
93 source: Option<SpaSource>,
94 assets: Option<EmbeddedAssets>,
95) -> Option<SpaSource> {
96 let Some(assets) = assets else {
97 return source;
98 };
99 let overlay_root = match source.as_ref() {
100 Some(existing) => existing.disk_root().to_string(),
101 None => assets.root().to_string(),
102 };
103 Some(SpaSource::Embedded {
104 assets,
105 overlay_root,
106 })
107}
108
109pub struct Host {
110 #[allow(dead_code)]
111 provider: Arc<ServiceProvider>,
112 dispatch_runtime: Arc<DispatchRuntime>,
113 pub options: AppOptions,
114 endpoints: Arc<dyn IEndpointCatalog>,
116 localizer: Arc<dyn IStringLocalizer>,
118 body_limits: BodyLimits,
120 pipeline: Arc<MiddlewarePipeline>,
121 #[allow(dead_code)]
123 router: Arc<Router>,
124 router_handler: HandlerFn,
126 mode: AppMode,
127 #[allow(dead_code)]
129 spa_source: Option<SpaSource>,
130 shutdown: Arc<tokio::sync::Notify>,
131 hosted_services: Vec<Arc<dyn IHostedService>>,
134}
135
136#[allow(clippy::type_complexity)]
137pub struct HostBuilder {
138 service_configs: Vec<Box<dyn FnOnce(ServiceCollection) -> ServiceCollection + Send>>,
139 mode: AppMode,
140 spa_source: Option<SpaSource>,
141 spa_disabled: bool,
142 options_modifiers: Vec<Box<dyn FnOnce(&mut AppOptions) + Send>>,
143 cors_config: Option<CorsConfig>,
144 auth_enabled: bool,
145 localization: Option<LocalizationOptions>,
146 health_registry: Arc<HealthCheckRegistry>,
147}
148
149#[allow(clippy::type_complexity)]
150pub struct HostAppBuilder {
151 options_modifiers: Vec<Box<dyn FnOnce(&mut AppOptions) + Send>>,
152}
153
154impl HostAppBuilder {
155 fn new() -> Self {
156 Self {
157 options_modifiers: Vec::new(),
158 }
159 }
160
161 #[allow(non_snake_case)]
162 pub fn useOptions<F>(&mut self, f: F)
163 where
164 F: FnOnce(&mut AppOptions) + Send + 'static,
165 {
166 self.options_modifiers.push(Box::new(f));
167 }
168}
169
170impl HostBuilder {
171 pub fn new() -> Self {
172 Self {
173 service_configs: Vec::new(),
174 mode: AppMode::from_env(),
177 spa_source: None,
178 spa_disabled: false,
179 options_modifiers: Vec::new(),
180 cors_config: None,
181 auth_enabled: false,
182 localization: None,
183 health_registry: HealthCheckRegistry::new(),
184 }
185 }
186
187 pub fn register<F>(mut self, f: F) -> Self
188 where
189 F: FnOnce(ServiceCollection) -> ServiceCollection + Send + 'static,
190 {
191 self.service_configs.push(Box::new(f));
192 self
193 }
194
195 pub fn configure<F>(mut self, f: F) -> Self
196 where
197 F: FnOnce(&mut HostAppBuilder) + Send + 'static,
198 {
199 let mut builder = HostAppBuilder::new();
200 f(&mut builder);
201 self.options_modifiers
202 .append(&mut builder.options_modifiers);
203 self
204 }
205
206 pub fn mode(mut self, mode: AppMode) -> Self {
207 self.mode = mode;
208 self
209 }
210
211 pub fn add_options<T>(self, section: &str) -> Self
223 where
224 T: for<'de> serde::Deserialize<'de> + Default + Send + Sync + 'static,
225 {
226 let mode = self.mode;
227 let section = section.to_string();
228 self.register(move |svc| {
229 let appsettings = config::load_appsettings(mode).unwrap_or_default();
230 let bound: T = config::bind_config(&appsettings, §ion);
231 tracing::info!(
232 "[Host] add_options<{}>(\"{}\") registered",
233 std::any::type_name::<T>(),
234 section
235 );
236 svc.instance(Arc::new(bound))
237 })
238 }
239
240 pub fn use_spa(mut self, source: impl Into<SpaSource>) -> Self {
255 self.spa_source = Some(match (self.spa_source.take(), source.into()) {
256 (Some(SpaSource::Embedded { assets, .. }), SpaSource::Disk(root)) => {
259 SpaSource::Embedded {
260 assets,
261 overlay_root: root,
262 }
263 }
264 (_, incoming) => incoming,
265 });
266 self
267 }
268
269 pub fn no_spa(mut self) -> Self {
275 self.spa_disabled = true;
276 self
277 }
278
279 pub fn use_cors(mut self, config: CorsConfig) -> Self {
280 self.cors_config = Some(config);
281 self
282 }
283
284 pub fn add_authentication(mut self) -> Self {
298 self.auth_enabled = true;
299 self
300 }
301
302 pub fn add_localization(mut self) -> Self {
318 self.localization = Some(LocalizationOptions::new());
319 self
320 }
321
322 pub fn add_localization_with(mut self, options: LocalizationOptions) -> Self {
338 self.localization = Some(options);
339 self
340 }
341
342 #[deprecated(
348 since = "0.6.0",
349 note = "Route authorization is enforced from endpoint metadata; remove this call"
350 )]
351 pub fn use_resource_authorization(self) -> Self {
352 self
353 }
354
355 pub fn add_memory_cache(mut self) -> Self {
367 let cache = Arc::new(MemoryCache::new());
368 self.service_configs
369 .push(Box::new(move |svc| svc.instance(cache)));
370 self
371 }
372
373 pub fn add_memory_cache_with(mut self, max_entries: usize) -> Self {
375 let cache = Arc::new(MemoryCache::new().with_max_entries(max_entries));
376 self.service_configs
377 .push(Box::new(move |svc| svc.instance(Arc::clone(&cache))));
378 self
379 }
380
381 pub fn use_middleware<T: IMiddleware + Default + 'static>(mut self) -> Self {
393 self.service_configs.push(Box::new(|svc| {
394 svc.singleton::<dyn IMiddleware>(|_| Arc::new(T::default()) as Arc<dyn IMiddleware>)
395 }));
396 self
397 }
398
399 pub fn use_middleware_with<F>(mut self, factory: F) -> Self
411 where
412 F: Fn() -> Arc<dyn IMiddleware> + Send + Sync + 'static,
413 {
414 self.service_configs.push(Box::new(move |svc| {
415 svc.singleton::<dyn IMiddleware>(move |_| factory())
416 }));
417 self
418 }
419
420 pub fn add_health_check<F>(self, name: impl Into<String>, check: F) -> Self
433 where
434 F: Fn() -> HealthStatus + Send + Sync + 'static,
435 {
436 self.health_registry.register(name, Arc::new(check));
437 self
438 }
439
440 pub fn build(self) -> Host {
441 let env_filter = std::env::var("RUST_LOG").unwrap_or_else(|_| "info".into());
444 if self.mode == AppMode::Development {
445 let _ = tracing_subscriber::fmt()
446 .with_env_filter(env_filter)
447 .without_time()
448 .with_target(false)
449 .try_init();
450 } else {
451 let _ = tracing_subscriber::fmt()
452 .json()
453 .with_env_filter(env_filter)
454 .try_init();
455 }
456
457 let appsettings =
460 config::load_appsettings(self.mode).unwrap_or_else(|| serde_json::json!({}));
461 let mut options: AppOptions = config::bind_root(&appsettings);
462 for modifier in self.options_modifiers {
463 modifier(&mut options);
464 }
465
466 let default_culture = if options.culture.default.trim().is_empty() {
467 "en".to_string()
468 } else {
469 options.culture.default.clone()
470 };
471 let localization = self.localization.unwrap_or_else(LocalizationOptions::none);
472 let localizer: Arc<dyn IStringLocalizer> = Arc::new(
473 StringLocalizer::new(default_culture, &localization)
474 .unwrap_or_else(|error| panic!("Invalid localization resources: {error}")),
475 );
476 let endpoints: Arc<dyn IEndpointCatalog> = Arc::new(EndpointCatalog::from_inventory());
477
478 let mut svc = ServiceCollection::from_injected();
479 for cfg in self.service_configs {
480 svc = cfg(svc);
481 }
482
483 let health_registry = Arc::clone(&self.health_registry);
485 svc = svc.instance(health_registry);
486
487 let endpoints_for_di = Arc::clone(&endpoints);
489 svc = svc.singleton::<dyn IEndpointCatalog>(move |_| Arc::clone(&endpoints_for_di));
490 let localizer_for_di = Arc::clone(&localizer);
491 svc = svc.singleton::<dyn IStringLocalizer>(move |_| Arc::clone(&localizer_for_di));
492
493 let provider = svc.build().unwrap_or_else(|e| {
494 panic!(
495 "Failed to build ServiceProvider: {}. Check your DI registrations.",
496 e
497 )
498 });
499
500 let handler_cache = Arc::new(webx_core::route::scan::HandlerCache::build());
502 let dispatch_runtime = Arc::new(DispatchRuntime::new(Arc::clone(&provider), handler_cache));
503
504 let body_limits = BodyLimits::from_options(&options);
506 body_limits.apply_spool_root();
507 tracing::info!(
508 "[Host] Uploads: max request {} bytes, max file {} bytes, spool >{} bytes to {}",
509 body_limits.multipart_body,
510 body_limits.file,
511 body_limits.memory_threshold,
512 body_limits.spool_dir.display()
513 );
514
515 let http_metrics = HttpMetrics::new();
516
517 let mut pipeline = MiddlewarePipeline::new();
518
519 pipeline.add_middleware(Arc::new(
522 crate::security_headers::SecurityHeadersMiddleware::new(),
523 ));
524 pipeline.add_middleware(Arc::new(crate::request_id::RequestIdMiddleware::new()));
525
526 if options.rate_limit.enabled {
527 tracing::info!(
528 "[Host] RateLimit enabled: {} req/s, burst {}, max_ips {}",
529 options.rate_limit.requests_per_second,
530 options.rate_limit.burst_size,
531 options.rate_limit.max_tracked_ips
532 );
533 pipeline.add_middleware(Arc::new(RateLimitMiddleware::from_config(
534 &options.rate_limit,
535 )));
536 }
537
538 if options.metrics.enabled {
539 tracing::info!("[Host] Metrics enabled at GET /metrics");
540 pipeline.add_middleware(Arc::new(MetricsMiddleware::new(Arc::clone(&http_metrics))));
541 }
542
543 let middlewares: Vec<Arc<dyn IMiddleware>> = provider.get_all::<dyn IMiddleware>();
544 for mw in middlewares {
545 pipeline.add_middleware(mw);
546 }
547
548 let cors = self.cors_config.unwrap_or_else(|| {
549 let cs = &options.cors;
550 CorsConfig {
551 origins: cs.origins.clone(),
552 methods: cs.methods.clone(),
553 headers: cs.headers.clone(),
554 expose_headers: cs.expose_headers.clone(),
555 allow_credentials: cs.allow_credentials,
556 max_age: cs.max_age,
557 }
558 });
559 pipeline.add_middleware(Arc::new(CorsMiddleware::new(cors)));
560
561 if self.auth_enabled {
562 let secret = options.jwt.secret.clone();
563 if !secret.is_empty() {
564 let jwt_auth = Arc::new(JwtAuth::new(
565 DecodingKey::from_secret(secret.as_bytes()),
566 Validation::default(),
567 ));
568 pipeline.add_middleware(Arc::new(jwt_middleware(jwt_auth)));
569 init_jwt_secret(&secret);
570 } else {
571 tracing::warn!(
572 "add_authentication() enabled but no JWT secret configured. Set Jwt.Secret in appsettings.json, JWT_SECRET, or APP__Jwt__Secret env var."
573 );
574 }
575 }
576
577 let spa_source = if self.spa_disabled {
585 None
586 } else {
587 let base = self.spa_source.clone().or_else(|| {
588 let candidate = webx_core::paths::app_base().join("wwwroot");
589 if candidate.is_dir() {
590 tracing::info!("[Host] Auto-detected SPA root: {}", candidate.display());
591 Some(SpaSource::Disk(candidate.to_string_lossy().into_owned()))
592 } else {
593 None
594 }
595 });
596 with_embedded_assets(base, EmbeddedAssets::registered().copied())
597 };
598 if let Some(ref source) = spa_source {
599 let spa = SpaMiddleware::from_source(source.clone());
600 log_spa_source(&spa);
601 pipeline.add_middleware(Arc::new(spa));
602 }
603
604 let pipeline = Arc::new(pipeline);
609
610 if self.mode == AppMode::Production
612 && self.auth_enabled
613 && is_weak_jwt_secret(&options.jwt.secret)
614 {
615 panic!(
616 "Production requires a strong JWT secret. Set APP__Jwt__Secret or JWT_SECRET env var (min 32 chars, not a placeholder)."
617 );
618 }
619
620 if self.mode == AppMode::Production && options.cors.origins.iter().any(|o| o == "*") {
622 panic!(
623 "Production forbids CORS origin '*'. Set explicit origins via Cors.Origins or APP__Cors__Origins."
624 );
625 }
626
627 let mut router = Router::new();
628 let mut route_count = 0usize;
629
630 crate::diagnostics::assert_route_configuration_valid();
631
632 let mut dispatch_map: std::collections::HashMap<
634 &'static str,
635 webx_core::route::scan::RouteDispatchFn,
636 > = std::collections::HashMap::new();
637
638 for dispatch in inventory::iter::<webx_core::route::scan::RouteDispatch> {
639 dispatch_map.insert(dispatch.handler_type, dispatch.dispatch);
640 }
641
642 for entry in inventory::iter::<RouteEntry> {
643 route_count += 1;
644 let stub = Arc::new(StubEndpoint {
645 method: entry.method.as_str(),
646 path: entry.path,
647 handler_type: entry.handler_type,
648 dispatch_fn: dispatch_map.get(entry.handler_type).copied(),
649 authorization: entry.authorization,
650 authorizers: collect_authorizers(provider.as_ref()),
651 });
652 router.register(entry.method, entry.path, stub);
653 }
654
655 let openapi_spec = generate_openapi_spec(
656 "Rust WebX API",
657 env!("CARGO_PKG_VERSION"),
658 localizer.as_ref(),
659 );
660 if self.mode == AppMode::Development {
661 let openapi_bytes = serde_json::to_vec(&openapi_spec).unwrap_or_default();
662 router.register(
663 HttpMethod::Get,
664 "/api/openapi.json",
665 Arc::new(StaticJsonEndpoint::new(openapi_bytes)),
666 );
667 router.register(
668 HttpMethod::Get,
669 "/api/openapi.html",
670 Arc::new(StaticHtmlEndpoint { body: APIUI_HTML }),
671 );
672 }
673
674 let health_registry = Arc::clone(&self.health_registry);
677 let health_endpoint: Arc<dyn IEndpoint> =
678 Arc::new(HealthCheckEndpoint::new(Arc::clone(&health_registry)));
679 router.register(HttpMethod::Get, "/health", Arc::clone(&health_endpoint));
680 router.register(HttpMethod::Get, "/healthz", health_endpoint);
681
682 let live_body =
684 serde_json::to_vec(&serde_json::json!({"status":"pass"})).unwrap_or_default();
685 router.register(
686 HttpMethod::Get,
687 "/health/live",
688 Arc::new(StaticJsonEndpoint {
689 body: live_body,
690 content_type: "application/health+json",
691 }),
692 );
693
694 router.register(
696 HttpMethod::Get,
697 "/health/ready",
698 Arc::new(HealthCheckEndpoint::new(health_registry)),
699 );
700
701 if options.metrics.enabled {
702 router.register(
703 HttpMethod::Get,
704 "/metrics",
705 Arc::new(MetricsEndpoint::new(http_metrics)),
706 );
707 }
708
709 if self.mode == AppMode::Development {
710 let version = env!("CARGO_PKG_VERSION");
711 tracing::info!("");
712 tracing::info!(" ----------------------------------------------------------------");
713 tracing::info!(" Rust WebX Framework v{}", version);
714 tracing::info!(" ----------------------------------------------------------------");
715 tracing::info!(" App: {}", options.app.name);
716 tracing::info!(" CORS: enabled");
717 if let Some(source) = spa_source.as_ref() {
718 tracing::info!(" SPA Root: {}", source.disk_root());
719 if let Some(assets) = source.embedded() {
720 tracing::info!(
721 " Embedded: {} files, {:.1} MiB from {:.1} MiB (built from {})",
722 assets.len(),
723 assets.total_bytes() as f64 / (1024.0 * 1024.0),
724 assets.total_raw_bytes() as f64 / (1024.0 * 1024.0),
725 assets.root(),
726 );
727 }
728 }
729 if route_count > 0 {
730 tracing::info!(" Routes: {} registered", route_count);
731 }
732 crate::diagnostics::log_startup_diagnostics();
733 let banner_urls = if options.app.urls.is_empty() {
734 vec!["http://localhost:5000".to_string()]
735 } else {
736 options
737 .app
738 .urls
739 .iter()
740 .map(|u| u.replace("0.0.0.0", "localhost"))
741 .collect::<Vec<_>>()
742 };
743 for url in &banner_urls {
744 tracing::info!(" OpenAPI: {}/api/openapi.html", url);
745 tracing::info!(" OpenAPI: {}/api/openapi.json", url);
746 }
747 tracing::info!(" ----------------------------------------------------------------");
748 tracing::info!("");
749 } else if route_count > 0 {
750 tracing::info!("{} route(s) registered", route_count);
751 }
752
753 let router = Arc::new(router);
754 let router_handler = make_router_handler(Arc::clone(&router));
755
756 let hosted_services: Vec<Arc<dyn IHostedService>> =
759 provider.get_all::<dyn IHostedService>();
760
761 Host {
762 provider,
763 dispatch_runtime,
764 options,
765 endpoints,
766 localizer,
767 body_limits,
768 pipeline,
769 router,
770 router_handler,
771 mode: self.mode,
772 spa_source,
773 shutdown: Arc::new(tokio::sync::Notify::new()),
774 hosted_services,
775 }
776 }
777}
778
779pub struct ServerHandle {
781 shutdown: Arc<tokio::sync::Notify>,
782}
783
784impl ServerHandle {
785 pub fn shutdown(self) {
788 self.shutdown.notify_waiters();
789 }
790}
791
792pub struct Server {
804 host: Host,
805}
806
807impl Server {
808 pub async fn run(self) -> Result<()> {
810 self.host.run().await
811 }
812
813 pub async fn run_at(self, addr: &str) -> Result<()> {
815 self.host.run_at(addr).await
816 }
817
818 pub fn handle(&self) -> ServerHandle {
820 self.host.server_handle()
821 }
822}
823
824impl Default for HostBuilder {
825 fn default() -> Self {
826 Self::new()
827 }
828}
829
830impl Host {
831 pub fn builder() -> HostBuilder {
832 HostBuilder::new()
833 }
834
835 pub fn options(&self) -> &AppOptions {
836 &self.options
837 }
838
839 pub fn provider(&self) -> &Arc<ServiceProvider> {
841 &self.provider
842 }
843
844 pub fn dispatch_runtime(&self) -> &Arc<DispatchRuntime> {
846 &self.dispatch_runtime
847 }
848
849 pub fn endpoints(&self) -> &dyn IEndpointCatalog {
855 &*self.endpoints
856 }
857
858 pub fn localizer(&self) -> &dyn IStringLocalizer {
860 &*self.localizer
861 }
862
863 pub async fn run(&self) -> Result<()> {
877 start_hosted_services(&self.hosted_services, &self.dispatch_runtime).await?;
878
879 let urls = if self.options.app.urls.is_empty() {
880 vec!["http://0.0.0.0:5000".to_string()]
881 } else {
882 self.options.app.urls.clone()
883 };
884
885 let mut http_addrs: Vec<String> = Vec::new();
886 let mut https_addrs: Vec<String> = Vec::new();
887
888 for url in &urls {
889 let (scheme, addr) = parse_url(url)?;
890 match scheme {
891 "http" => http_addrs.push(addr),
892 "https" => https_addrs.push(addr),
893 other => {
894 return Err(webx_core::error::Error::Http(format!(
895 "Unsupported URL scheme '{}' in '{}'",
896 other, url
897 )))
898 }
899 }
900 }
901
902 let acceptor = if !https_addrs.is_empty() {
903 let tls = &self.options.tls;
904 if tls.cert_path.is_empty() || tls.key_path.is_empty() {
905 return Err(webx_core::error::Error::Http(
906 "HTTPS URLs require Tls.CertPath and Tls.KeyPath".into(),
907 ));
908 }
909 Some(build_tls_acceptor(&tls.cert_path, &tls.key_path)?)
910 } else {
911 None
912 };
913
914 if self.mode == AppMode::Development {
916 tracing::info!("");
917 for url in &urls {
918 let display_url = url.replace("0.0.0.0", "localhost");
919 tracing::info!(" Listening on {}", display_url);
920 }
921 } else {
922 tracing::info!("Listening on {} url(s)", urls.len());
923 }
924
925 let notify = Arc::clone(&self.shutdown);
926 install_shutdown_handler(Arc::clone(¬ify));
927 spawn_async_shutdown_listener(Arc::clone(¬ify));
928
929 let mut handles = Vec::new();
930 let pipeline = Arc::clone(&self.pipeline);
931 let router_handler = self.router_handler.clone();
932 let dispatch_runtime = Arc::clone(&self.dispatch_runtime);
933 let mode = self.mode;
934 let body_limits = self.body_limits.clone();
935 let max_connections = self.options.app.max_connections;
936
937 for addr in &http_addrs {
938 let addr = addr.clone();
939 let n = std::sync::Arc::clone(¬ify);
940 let p = Arc::clone(&pipeline);
941 let rh = router_handler.clone();
942 let dr = Arc::clone(&dispatch_runtime);
943 handles.push(tokio::spawn(serve_http(
944 addr,
945 n,
946 p,
947 rh,
948 dr,
949 mode,
950 body_limits.clone(),
951 max_connections,
952 )));
953 }
954
955 if let Some(ref tls_acceptor) = acceptor {
956 for addr in &https_addrs {
957 let addr = addr.clone();
958 let n = std::sync::Arc::clone(¬ify);
959 let p = Arc::clone(&pipeline);
960 let rh = router_handler.clone();
961 let dr = Arc::clone(&dispatch_runtime);
962 let a = tls_acceptor.clone();
963 handles.push(tokio::spawn(serve_https(
964 addr,
965 a,
966 n,
967 p,
968 rh,
969 dr,
970 mode,
971 body_limits.clone(),
972 max_connections,
973 )));
974 }
975 }
976
977 for h in handles {
979 let _ = h.await;
980 }
981
982 stop_hosted_services(&self.hosted_services).await;
983
984 Ok(())
985 }
986
987 pub async fn run_at(&self, addr: &str) -> Result<()> {
989 start_hosted_services(&self.hosted_services, &self.dispatch_runtime).await?;
990
991 let notify = Arc::clone(&self.shutdown);
992 install_shutdown_handler(Arc::clone(¬ify));
993 spawn_async_shutdown_listener(Arc::clone(¬ify));
994 serve_http(
995 addr.to_string(),
996 notify,
997 Arc::clone(&self.pipeline),
998 self.router_handler.clone(),
999 Arc::clone(&self.dispatch_runtime),
1000 self.mode,
1001 self.body_limits.clone(),
1002 self.options.app.max_connections,
1003 )
1004 .await;
1005
1006 stop_hosted_services(&self.hosted_services).await;
1007 Ok(())
1008 }
1009
1010 pub fn server_handle(&self) -> ServerHandle {
1021 ServerHandle {
1022 shutdown: Arc::clone(&self.shutdown),
1023 }
1024 }
1025
1026 pub fn into_server(self) -> Server {
1031 Server { host: self }
1032 }
1033}
1034
1035#[async_trait::async_trait]
1036impl IHost for Host {
1037 async fn run(&self, addr: &str) -> Result<()> {
1038 self.run_at(addr).await
1039 }
1040 async fn stop(&self) -> Result<()> {
1041 tracing::info!("Stop requested.");
1042 self.shutdown.notify_waiters();
1043 Ok(())
1044 }
1045}
1046
1047fn make_router_handler(router: Arc<Router>) -> HandlerFn {
1048 Arc::new(move |ctx: &mut dyn IHttpContext| {
1049 let router = Arc::clone(&router);
1050 Box::pin(async move {
1051 match router.match_route(ctx).await? {
1052 Some((endpoint, params, pattern)) => {
1053 drop(router);
1054 for (key, value) in params {
1055 ctx.request_mut().route_params_mut().insert(key, value);
1056 }
1057 *ctx.request_mut().route_pattern_mut() = Some(pattern);
1058 endpoint.handle(ctx).await
1059 }
1060 None => {
1061 let path_exists = router.path_exists(ctx.request().path());
1062 drop(router);
1063 if !ctx.response().has_body() {
1066 if path_exists {
1067 write_error_response(ctx, 405, "Method Not Allowed").await;
1068 } else {
1069 write_error_response(ctx, 404, "Not Found").await;
1070 }
1071 }
1072 Ok(())
1073 }
1074 }
1075 })
1076 })
1077}
1078
1079async fn handle_request(
1080 req: Request<Incoming>,
1081 pipeline: Arc<MiddlewarePipeline>,
1082 router_handler: HandlerFn,
1083 dispatch_runtime: Arc<DispatchRuntime>,
1084 body_limits: BodyLimits,
1085) -> std::result::Result<hyper::Response<RespBody>, std::convert::Infallible> {
1086 let mut ctx = HttpContext::new(req, body_limits);
1087
1088 if ctx.response().status() < 400 {
1091 let result = dispatch_runtime
1092 .run(async { pipeline.execute(&mut ctx, router_handler).await })
1093 .await;
1094 if let Err(e) = result {
1095 let status = e.status_code();
1096 write_error_response(&mut ctx, status, &e.to_string()).await;
1097 }
1098 }
1099
1100 Ok(ctx.into_response().await)
1101}
1102
1103async fn write_error_response(ctx: &mut dyn IHttpContext, status: u16, message: &str) {
1104 write_problem_response(ctx, build_problem(status, message)).await;
1105}
1106
1107fn install_shutdown_handler(shutdown: Arc<tokio::sync::Notify>) {
1117 use std::sync::atomic::{AtomicUsize, Ordering};
1118
1119 static CTRL_COUNT: AtomicUsize = AtomicUsize::new(0);
1120
1121 if let Err(e) = ctrlc::set_handler(move || {
1122 let n = CTRL_COUNT.fetch_add(1, Ordering::SeqCst);
1123 if n == 0 {
1124 tracing::info!("Shutdown signal received, draining connections...");
1125 shutdown.notify_waiters();
1126 } else {
1127 tracing::warn!("Second interrupt — force exiting.");
1128 std::process::exit(130);
1129 }
1130 }) {
1131 tracing::warn!("Could not install Ctrl+C handler: {}", e);
1132 }
1133}
1134
1135fn spawn_async_shutdown_listener(shutdown: Arc<tokio::sync::Notify>) {
1137 tokio::spawn(async move {
1138 wait_for_shutdown_signal().await;
1139 tracing::info!("Async shutdown signal received, draining connections...");
1140 shutdown.notify_waiters();
1141 });
1142}
1143
1144async fn wait_for_shutdown_signal() {
1145 #[cfg(unix)]
1146 {
1147 use tokio::signal::unix::{signal, SignalKind};
1148 let mut sigterm = signal(SignalKind::terminate()).expect("install SIGTERM handler");
1149 tokio::select! {
1150 res = tokio::signal::ctrl_c() => {
1151 if let Err(e) = res {
1152 tracing::warn!("Ctrl+C listener error: {}", e);
1153 }
1154 }
1155 _ = sigterm.recv() => {}
1156 }
1157 }
1158 #[cfg(not(unix))]
1159 {
1160 if let Err(e) = tokio::signal::ctrl_c().await {
1161 tracing::warn!("Ctrl+C listener error: {}", e);
1162 }
1163 }
1164}
1165
1166async fn start_hosted_services(
1167 services: &[Arc<dyn IHostedService>],
1168 dispatch_runtime: &DispatchRuntime,
1169) -> Result<()> {
1170 if services.is_empty() {
1171 return Ok(());
1172 }
1173 tracing::info!("Starting {} hosted service(s)...", services.len());
1174 dispatch_runtime
1175 .run(async {
1176 for svc in services {
1177 svc.start().await?;
1178 }
1179 Ok::<(), webx_core::error::Error>(())
1180 })
1181 .await?;
1182 tracing::info!("All hosted services started.");
1183 Ok(())
1184}
1185
1186async fn stop_hosted_services(services: &[Arc<dyn IHostedService>]) {
1187 if services.is_empty() {
1188 return;
1189 }
1190 tracing::info!("Stopping {} hosted service(s)...", services.len());
1191 for svc in services {
1192 if let Err(e) = svc.stop().await {
1193 tracing::warn!("Hosted service stop error: {}", e);
1194 }
1195 }
1196 tracing::info!("All hosted services stopped.");
1197}
1198
1199async fn drain_connections(join_set: &mut JoinSet<()>, label: &str, mode: AppMode) {
1200 let timeout_secs = if mode == AppMode::Development { 5 } else { 30 };
1201 let drain_future = async {
1202 while let Some(result) = join_set.join_next().await {
1203 if let Err(e) = result {
1204 tracing::error!("Connection task panicked: {:?}", e);
1205 }
1206 }
1207 };
1208
1209 match tokio::time::timeout(std::time::Duration::from_secs(timeout_secs), drain_future).await {
1210 Ok(_) => tracing::info!("{}: drained.", label),
1211 Err(_) => {
1212 tracing::warn!("{}: drain timeout, aborting connections.", label);
1213 join_set.abort_all();
1214 while join_set.join_next().await.is_some() {}
1215 }
1216 }
1217}
1218
1219fn parse_url(url: &str) -> Result<(&str, String)> {
1226 if let Some(rest) = url.strip_prefix("https://") {
1227 Ok(("https", rest.to_string()))
1228 } else if let Some(rest) = url.strip_prefix("http://") {
1229 Ok(("http", rest.to_string()))
1230 } else {
1231 Err(webx_core::error::Error::Http(format!(
1232 "Invalid URL '{}'. Use http://host:port or https://host:port",
1233 url
1234 )))
1235 }
1236}
1237
1238#[allow(clippy::too_many_arguments)]
1240async fn serve_http(
1241 addr: String,
1242 shutdown: std::sync::Arc<tokio::sync::Notify>,
1243 pipeline: Arc<MiddlewarePipeline>,
1244 router_handler: HandlerFn,
1245 dispatch_runtime: Arc<DispatchRuntime>,
1246 mode: AppMode,
1247 body_limits: BodyLimits,
1248 max_connections: usize,
1249) {
1250 let listener = match tokio::net::TcpListener::bind(&addr).await {
1251 Ok(l) => l,
1252 Err(e) => {
1253 tracing::error!("Failed to bind HTTP on {}: {}", addr, e);
1254 return;
1255 }
1256 };
1257
1258 let sem = Arc::new(tokio::sync::Semaphore::new(max_connections));
1259 let mut join_set = JoinSet::new();
1260
1261 loop {
1262 tokio::select! {
1263 accept_result = listener.accept() => {
1264 let (stream, _) = match accept_result {
1265 Ok(v) => v,
1266 Err(e) => {
1267 tracing::error!("Accept error (will retry): {}", e);
1268 tokio::time::sleep(std::time::Duration::from_millis(100)).await;
1269 continue;
1270 }
1271 };
1272
1273 while join_set.try_join_next().is_some() {}
1274
1275 let permit = match sem.clone().acquire_owned().await {
1276 Ok(p) => p,
1277 Err(_) => continue,
1278 };
1279
1280 let io = TokioIo::new(stream);
1281 let pipeline = Arc::clone(&pipeline);
1282 let router_handler = router_handler.clone();
1283 let dispatch_runtime = Arc::clone(&dispatch_runtime);
1284 let body_limits = body_limits.clone();
1285
1286 join_set.spawn(async move {
1287 let _permit = permit;
1288 let svc_fn = service_fn(move |req: Request<Incoming>| {
1289 let pipeline = Arc::clone(&pipeline);
1290 let router_handler = router_handler.clone();
1291 let dispatch_runtime = Arc::clone(&dispatch_runtime);
1292 let body_limits = body_limits.clone();
1293 let mode = mode;
1294 async move {
1295 let start = Instant::now();
1296 let method = req.method().to_string();
1297 let path = req.uri().path().to_string();
1298 let result = handle_request(
1299 req,
1300 pipeline,
1301 router_handler,
1302 dispatch_runtime,
1303 body_limits,
1304 )
1305 .await;
1306 let elapsed = start.elapsed();
1307 if mode == AppMode::Development {
1308 let status =
1309 result.as_ref().map(|r| r.status().as_u16()).unwrap_or(500);
1310 tracing::info!(
1311 "[{}] {} -> {} ({:.0}ms)",
1312 method,
1313 path,
1314 status,
1315 elapsed.as_secs_f64() * 1000.0
1316 );
1317 }
1318 result
1319 }
1320 });
1321
1322 if let Err(err) = hyper::server::conn::http1::Builder::new()
1323 .serve_connection(io, svc_fn)
1324 .await
1325 {
1326 tracing::error!("Connection error: {}", err);
1327 }
1328 });
1329 }
1330 _ = shutdown.notified() => {
1331 tracing::info!("HTTP {}: stop accepting, draining...", addr);
1332 break;
1333 }
1334 }
1335 }
1336
1337 drain_connections(&mut join_set, &format!("HTTP {}", addr), mode).await;
1338}
1339
1340#[allow(clippy::too_many_arguments)]
1342async fn serve_https(
1343 addr: String,
1344 acceptor: TlsAcceptor,
1345 shutdown: std::sync::Arc<tokio::sync::Notify>,
1346 pipeline: Arc<MiddlewarePipeline>,
1347 router_handler: HandlerFn,
1348 dispatch_runtime: Arc<DispatchRuntime>,
1349 mode: AppMode,
1350 body_limits: BodyLimits,
1351 max_connections: usize,
1352) {
1353 let listener = match tokio::net::TcpListener::bind(&addr).await {
1354 Ok(l) => l,
1355 Err(e) => {
1356 tracing::error!("Failed to bind HTTPS on {}: {}", addr, e);
1357 return;
1358 }
1359 };
1360
1361 let sem = Arc::new(tokio::sync::Semaphore::new(max_connections));
1362 let mut join_set = JoinSet::new();
1363
1364 loop {
1365 tokio::select! {
1366 accept_result = listener.accept() => {
1367 let (stream, _) = match accept_result {
1368 Ok(v) => v,
1369 Err(e) => {
1370 tracing::error!("Accept error (will retry): {}", e);
1371 tokio::time::sleep(std::time::Duration::from_millis(100)).await;
1372 continue;
1373 }
1374 };
1375
1376 while join_set.try_join_next().is_some() {}
1377
1378 let permit = match sem.clone().acquire_owned().await {
1379 Ok(p) => p,
1380 Err(_) => continue,
1381 };
1382
1383 let acceptor = acceptor.clone();
1384 let pipeline = Arc::clone(&pipeline);
1385 let router_handler = router_handler.clone();
1386 let dispatch_runtime = Arc::clone(&dispatch_runtime);
1387 let body_limits = body_limits.clone();
1388
1389 join_set.spawn(async move {
1390 let _permit = permit;
1391 match acceptor.accept(stream).await {
1392 Ok(tls_stream) => {
1393 let io = TokioIo::new(tls_stream);
1394 let svc_fn = service_fn(move |req: Request<Incoming>| {
1395 let pipeline = Arc::clone(&pipeline);
1396 let router_handler = router_handler.clone();
1397 let dispatch_runtime = Arc::clone(&dispatch_runtime);
1398 let body_limits = body_limits.clone();
1399 let mode = mode;
1400 async move {
1401 let start = Instant::now();
1402 let method = req.method().to_string();
1403 let path = req.uri().path().to_string();
1404 let result = handle_request(
1405 req,
1406 pipeline,
1407 router_handler,
1408 dispatch_runtime,
1409 body_limits,
1410 )
1411 .await;
1412 let elapsed = start.elapsed();
1413 if mode == AppMode::Development {
1414 let status =
1415 result.as_ref().map(|r| r.status().as_u16()).unwrap_or(500);
1416 tracing::info!(
1417 "[{}] {} -> {} ({:.0}ms)",
1418 method,
1419 path,
1420 status,
1421 elapsed.as_secs_f64() * 1000.0
1422 );
1423 }
1424 result
1425 }
1426 });
1427
1428 if let Err(err) = hyper::server::conn::http1::Builder::new()
1429 .serve_connection(io, svc_fn)
1430 .await
1431 {
1432 tracing::error!("TLS connection error: {}", err);
1433 }
1434 }
1435 Err(e) => {
1436 tracing::error!("TLS handshake error: {}", e);
1437 }
1438 }
1439 });
1440 }
1441 _ = shutdown.notified() => {
1442 tracing::info!("HTTPS {}: stop accepting, draining...", addr);
1443 break;
1444 }
1445 }
1446 }
1447
1448 drain_connections(&mut join_set, &format!("HTTPS {}", addr), mode).await;
1449}
1450
1451fn build_tls_acceptor(cert_path: &str, key_path: &str) -> Result<TlsAcceptor> {
1457 use std::fs::File;
1458 use std::io::BufReader;
1459
1460 if cert_path.is_empty() || key_path.is_empty() {
1461 return Err(webx_core::error::Error::Http(
1462 "TLS certificate or key path not configured.".into(),
1463 ));
1464 }
1465
1466 let cert_file = File::open(cert_path).map_err(|e| {
1467 webx_core::error::Error::Http(format!("Cannot open cert '{}': {}", cert_path, e))
1468 })?;
1469 let mut cert_reader = BufReader::new(cert_file);
1470 let certs: Vec<CertificateDer> = certs(&mut cert_reader).filter_map(|r| r.ok()).collect();
1471 if certs.is_empty() {
1472 return Err(webx_core::error::Error::Http(format!(
1473 "No valid certs in '{}'",
1474 cert_path
1475 )));
1476 }
1477
1478 let key_file = File::open(key_path).map_err(|e| {
1479 webx_core::error::Error::Http(format!("Cannot open key '{}': {}", key_path, e))
1480 })?;
1481 let mut key_reader = BufReader::new(key_file);
1482 let key = pkcs8_private_keys(&mut key_reader)
1483 .filter_map(|r| r.ok())
1484 .map(PrivateKeyDer::from)
1485 .next()
1486 .or_else(|| {
1487 let key_file2 = File::open(key_path).map(BufReader::new).ok()?;
1488 let mut kr2 = key_file2;
1489 let rsa_keys: Vec<PrivateKeyDer> = rsa_private_keys(&mut kr2)
1490 .filter_map(|r| r.ok())
1491 .map(PrivateKeyDer::from)
1492 .collect();
1493 rsa_keys.into_iter().next()
1494 })
1495 .ok_or_else(|| {
1496 webx_core::error::Error::Http(format!("No valid private key in '{}'", key_path))
1497 })?;
1498
1499 let config = ServerConfig::builder()
1500 .with_no_client_auth()
1501 .with_single_cert(certs, key)
1502 .map_err(|e| webx_core::error::Error::Http(format!("TLS config error: {}", e)))?;
1503
1504 Ok(TlsAcceptor::from(std::sync::Arc::new(config)))
1505}
1506
1507#[cfg(test)]
1508mod embedded_attach_tests {
1509 use super::with_embedded_assets;
1510 use webx_spa::{ContentEncoding, EmbeddedAsset, EmbeddedAssets, SpaSource};
1511
1512 const FILES: &[EmbeddedAsset] = &[EmbeddedAsset {
1513 path: "index.html",
1514 bytes: b"hi",
1515 raw_len: 2,
1516 encoding: ContentEncoding::Identity,
1517 content_type: "text/html",
1518 etag: "\"x\"",
1519 }];
1520
1521 #[test]
1522 fn no_table_leaves_disk_source() {
1523 let source = Some(SpaSource::Disk("wwwroot".into()));
1524 let merged = with_embedded_assets(source, None);
1525 assert!(matches!(merged, Some(SpaSource::Disk(ref r)) if r == "wwwroot"));
1526 }
1527
1528 #[test]
1529 fn table_layers_under_use_spa_overlay() {
1530 let assets = EmbeddedAssets::new("../wwwroot", FILES);
1531 let merged = with_embedded_assets(Some(SpaSource::Disk("wwwroot".into())), Some(assets));
1532 match merged {
1533 Some(SpaSource::Embedded {
1534 overlay_root,
1535 assets,
1536 }) => {
1537 assert_eq!(overlay_root, "wwwroot");
1538 assert_eq!(assets.root(), "../wwwroot");
1539 }
1540 other => panic!("expected embedded, got {other:?}"),
1541 }
1542 }
1543
1544 #[test]
1545 fn table_alone_uses_build_root_as_overlay() {
1546 let assets = EmbeddedAssets::new("../wwwroot", FILES);
1547 let merged = with_embedded_assets(None, Some(assets));
1548 match merged {
1549 Some(SpaSource::Embedded { overlay_root, .. }) => {
1550 assert_eq!(overlay_root, "../wwwroot");
1551 }
1552 other => panic!("expected embedded, got {other:?}"),
1553 }
1554 }
1555
1556 #[test]
1557 fn host_crate_registers_no_asset_table() {
1558 assert!(EmbeddedAssets::registered().is_none());
1559 }
1560}