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 match e {
1096 webx_core::error::Error::Problem(problem) => {
1100 write_problem_response(&mut ctx, *problem).await
1101 }
1102 other => {
1103 let status = other.status_code();
1104 write_error_response(&mut ctx, status, &other.to_string()).await;
1105 }
1106 }
1107 }
1108 }
1109
1110 Ok(ctx.into_response().await)
1111}
1112
1113async fn write_error_response(ctx: &mut dyn IHttpContext, status: u16, message: &str) {
1114 write_problem_response(ctx, build_problem(status, message)).await;
1115}
1116
1117fn install_shutdown_handler(shutdown: Arc<tokio::sync::Notify>) {
1127 use std::sync::atomic::{AtomicUsize, Ordering};
1128
1129 static CTRL_COUNT: AtomicUsize = AtomicUsize::new(0);
1130
1131 if let Err(e) = ctrlc::set_handler(move || {
1132 let n = CTRL_COUNT.fetch_add(1, Ordering::SeqCst);
1133 if n == 0 {
1134 tracing::info!("Shutdown signal received, draining connections...");
1135 shutdown.notify_waiters();
1136 } else {
1137 tracing::warn!("Second interrupt — force exiting.");
1138 std::process::exit(130);
1139 }
1140 }) {
1141 tracing::warn!("Could not install Ctrl+C handler: {}", e);
1142 }
1143}
1144
1145fn spawn_async_shutdown_listener(shutdown: Arc<tokio::sync::Notify>) {
1147 tokio::spawn(async move {
1148 wait_for_shutdown_signal().await;
1149 tracing::info!("Async shutdown signal received, draining connections...");
1150 shutdown.notify_waiters();
1151 });
1152}
1153
1154async fn wait_for_shutdown_signal() {
1155 #[cfg(unix)]
1156 {
1157 use tokio::signal::unix::{signal, SignalKind};
1158 let mut sigterm = signal(SignalKind::terminate()).expect("install SIGTERM handler");
1159 tokio::select! {
1160 res = tokio::signal::ctrl_c() => {
1161 if let Err(e) = res {
1162 tracing::warn!("Ctrl+C listener error: {}", e);
1163 }
1164 }
1165 _ = sigterm.recv() => {}
1166 }
1167 }
1168 #[cfg(not(unix))]
1169 {
1170 if let Err(e) = tokio::signal::ctrl_c().await {
1171 tracing::warn!("Ctrl+C listener error: {}", e);
1172 }
1173 }
1174}
1175
1176async fn start_hosted_services(
1177 services: &[Arc<dyn IHostedService>],
1178 dispatch_runtime: &DispatchRuntime,
1179) -> Result<()> {
1180 if services.is_empty() {
1181 return Ok(());
1182 }
1183 tracing::info!("Starting {} hosted service(s)...", services.len());
1184 dispatch_runtime
1185 .run(async {
1186 for svc in services {
1187 svc.start().await?;
1188 }
1189 Ok::<(), webx_core::error::Error>(())
1190 })
1191 .await?;
1192 tracing::info!("All hosted services started.");
1193 Ok(())
1194}
1195
1196async fn stop_hosted_services(services: &[Arc<dyn IHostedService>]) {
1197 if services.is_empty() {
1198 return;
1199 }
1200 tracing::info!("Stopping {} hosted service(s)...", services.len());
1201 for svc in services {
1202 if let Err(e) = svc.stop().await {
1203 tracing::warn!("Hosted service stop error: {}", e);
1204 }
1205 }
1206 tracing::info!("All hosted services stopped.");
1207}
1208
1209async fn drain_connections(join_set: &mut JoinSet<()>, label: &str, mode: AppMode) {
1210 let timeout_secs = if mode == AppMode::Development { 5 } else { 30 };
1211 let drain_future = async {
1212 while let Some(result) = join_set.join_next().await {
1213 if let Err(e) = result {
1214 tracing::error!("Connection task panicked: {:?}", e);
1215 }
1216 }
1217 };
1218
1219 match tokio::time::timeout(std::time::Duration::from_secs(timeout_secs), drain_future).await {
1220 Ok(_) => tracing::info!("{}: drained.", label),
1221 Err(_) => {
1222 tracing::warn!("{}: drain timeout, aborting connections.", label);
1223 join_set.abort_all();
1224 while join_set.join_next().await.is_some() {}
1225 }
1226 }
1227}
1228
1229fn parse_url(url: &str) -> Result<(&str, String)> {
1236 if let Some(rest) = url.strip_prefix("https://") {
1237 Ok(("https", rest.to_string()))
1238 } else if let Some(rest) = url.strip_prefix("http://") {
1239 Ok(("http", rest.to_string()))
1240 } else {
1241 Err(webx_core::error::Error::Http(format!(
1242 "Invalid URL '{}'. Use http://host:port or https://host:port",
1243 url
1244 )))
1245 }
1246}
1247
1248#[allow(clippy::too_many_arguments)]
1250async fn serve_http(
1251 addr: String,
1252 shutdown: std::sync::Arc<tokio::sync::Notify>,
1253 pipeline: Arc<MiddlewarePipeline>,
1254 router_handler: HandlerFn,
1255 dispatch_runtime: Arc<DispatchRuntime>,
1256 mode: AppMode,
1257 body_limits: BodyLimits,
1258 max_connections: usize,
1259) {
1260 let listener = match tokio::net::TcpListener::bind(&addr).await {
1261 Ok(l) => l,
1262 Err(e) => {
1263 tracing::error!("Failed to bind HTTP on {}: {}", addr, e);
1264 return;
1265 }
1266 };
1267
1268 let sem = Arc::new(tokio::sync::Semaphore::new(max_connections));
1269 let mut join_set = JoinSet::new();
1270
1271 loop {
1272 tokio::select! {
1273 accept_result = listener.accept() => {
1274 let (stream, _) = match accept_result {
1275 Ok(v) => v,
1276 Err(e) => {
1277 tracing::error!("Accept error (will retry): {}", e);
1278 tokio::time::sleep(std::time::Duration::from_millis(100)).await;
1279 continue;
1280 }
1281 };
1282
1283 while join_set.try_join_next().is_some() {}
1284
1285 let permit = match sem.clone().acquire_owned().await {
1286 Ok(p) => p,
1287 Err(_) => continue,
1288 };
1289
1290 let io = TokioIo::new(stream);
1291 let pipeline = Arc::clone(&pipeline);
1292 let router_handler = router_handler.clone();
1293 let dispatch_runtime = Arc::clone(&dispatch_runtime);
1294 let body_limits = body_limits.clone();
1295
1296 join_set.spawn(async move {
1297 let _permit = permit;
1298 let svc_fn = service_fn(move |req: Request<Incoming>| {
1299 let pipeline = Arc::clone(&pipeline);
1300 let router_handler = router_handler.clone();
1301 let dispatch_runtime = Arc::clone(&dispatch_runtime);
1302 let body_limits = body_limits.clone();
1303 let mode = mode;
1304 async move {
1305 let start = Instant::now();
1306 let method = req.method().to_string();
1307 let path = req.uri().path().to_string();
1308 let result = handle_request(
1309 req,
1310 pipeline,
1311 router_handler,
1312 dispatch_runtime,
1313 body_limits,
1314 )
1315 .await;
1316 let elapsed = start.elapsed();
1317 if mode == AppMode::Development {
1318 let status =
1319 result.as_ref().map(|r| r.status().as_u16()).unwrap_or(500);
1320 tracing::info!(
1321 "[{}] {} -> {} ({:.0}ms)",
1322 method,
1323 path,
1324 status,
1325 elapsed.as_secs_f64() * 1000.0
1326 );
1327 }
1328 result
1329 }
1330 });
1331
1332 if let Err(err) = hyper::server::conn::http1::Builder::new()
1333 .serve_connection(io, svc_fn)
1334 .await
1335 {
1336 tracing::error!("Connection error: {}", err);
1337 }
1338 });
1339 }
1340 _ = shutdown.notified() => {
1341 tracing::info!("HTTP {}: stop accepting, draining...", addr);
1342 break;
1343 }
1344 }
1345 }
1346
1347 drain_connections(&mut join_set, &format!("HTTP {}", addr), mode).await;
1348}
1349
1350#[allow(clippy::too_many_arguments)]
1352async fn serve_https(
1353 addr: String,
1354 acceptor: TlsAcceptor,
1355 shutdown: std::sync::Arc<tokio::sync::Notify>,
1356 pipeline: Arc<MiddlewarePipeline>,
1357 router_handler: HandlerFn,
1358 dispatch_runtime: Arc<DispatchRuntime>,
1359 mode: AppMode,
1360 body_limits: BodyLimits,
1361 max_connections: usize,
1362) {
1363 let listener = match tokio::net::TcpListener::bind(&addr).await {
1364 Ok(l) => l,
1365 Err(e) => {
1366 tracing::error!("Failed to bind HTTPS on {}: {}", addr, e);
1367 return;
1368 }
1369 };
1370
1371 let sem = Arc::new(tokio::sync::Semaphore::new(max_connections));
1372 let mut join_set = JoinSet::new();
1373
1374 loop {
1375 tokio::select! {
1376 accept_result = listener.accept() => {
1377 let (stream, _) = match accept_result {
1378 Ok(v) => v,
1379 Err(e) => {
1380 tracing::error!("Accept error (will retry): {}", e);
1381 tokio::time::sleep(std::time::Duration::from_millis(100)).await;
1382 continue;
1383 }
1384 };
1385
1386 while join_set.try_join_next().is_some() {}
1387
1388 let permit = match sem.clone().acquire_owned().await {
1389 Ok(p) => p,
1390 Err(_) => continue,
1391 };
1392
1393 let acceptor = acceptor.clone();
1394 let pipeline = Arc::clone(&pipeline);
1395 let router_handler = router_handler.clone();
1396 let dispatch_runtime = Arc::clone(&dispatch_runtime);
1397 let body_limits = body_limits.clone();
1398
1399 join_set.spawn(async move {
1400 let _permit = permit;
1401 match acceptor.accept(stream).await {
1402 Ok(tls_stream) => {
1403 let io = TokioIo::new(tls_stream);
1404 let svc_fn = service_fn(move |req: Request<Incoming>| {
1405 let pipeline = Arc::clone(&pipeline);
1406 let router_handler = router_handler.clone();
1407 let dispatch_runtime = Arc::clone(&dispatch_runtime);
1408 let body_limits = body_limits.clone();
1409 let mode = mode;
1410 async move {
1411 let start = Instant::now();
1412 let method = req.method().to_string();
1413 let path = req.uri().path().to_string();
1414 let result = handle_request(
1415 req,
1416 pipeline,
1417 router_handler,
1418 dispatch_runtime,
1419 body_limits,
1420 )
1421 .await;
1422 let elapsed = start.elapsed();
1423 if mode == AppMode::Development {
1424 let status =
1425 result.as_ref().map(|r| r.status().as_u16()).unwrap_or(500);
1426 tracing::info!(
1427 "[{}] {} -> {} ({:.0}ms)",
1428 method,
1429 path,
1430 status,
1431 elapsed.as_secs_f64() * 1000.0
1432 );
1433 }
1434 result
1435 }
1436 });
1437
1438 if let Err(err) = hyper::server::conn::http1::Builder::new()
1439 .serve_connection(io, svc_fn)
1440 .await
1441 {
1442 tracing::error!("TLS connection error: {}", err);
1443 }
1444 }
1445 Err(e) => {
1446 tracing::error!("TLS handshake error: {}", e);
1447 }
1448 }
1449 });
1450 }
1451 _ = shutdown.notified() => {
1452 tracing::info!("HTTPS {}: stop accepting, draining...", addr);
1453 break;
1454 }
1455 }
1456 }
1457
1458 drain_connections(&mut join_set, &format!("HTTPS {}", addr), mode).await;
1459}
1460
1461fn build_tls_acceptor(cert_path: &str, key_path: &str) -> Result<TlsAcceptor> {
1467 use std::fs::File;
1468 use std::io::BufReader;
1469
1470 if cert_path.is_empty() || key_path.is_empty() {
1471 return Err(webx_core::error::Error::Http(
1472 "TLS certificate or key path not configured.".into(),
1473 ));
1474 }
1475
1476 let cert_file = File::open(cert_path).map_err(|e| {
1477 webx_core::error::Error::Http(format!("Cannot open cert '{}': {}", cert_path, e))
1478 })?;
1479 let mut cert_reader = BufReader::new(cert_file);
1480 let certs: Vec<CertificateDer> = certs(&mut cert_reader).filter_map(|r| r.ok()).collect();
1481 if certs.is_empty() {
1482 return Err(webx_core::error::Error::Http(format!(
1483 "No valid certs in '{}'",
1484 cert_path
1485 )));
1486 }
1487
1488 let key_file = File::open(key_path).map_err(|e| {
1489 webx_core::error::Error::Http(format!("Cannot open key '{}': {}", key_path, e))
1490 })?;
1491 let mut key_reader = BufReader::new(key_file);
1492 let key = pkcs8_private_keys(&mut key_reader)
1493 .filter_map(|r| r.ok())
1494 .map(PrivateKeyDer::from)
1495 .next()
1496 .or_else(|| {
1497 let key_file2 = File::open(key_path).map(BufReader::new).ok()?;
1498 let mut kr2 = key_file2;
1499 let rsa_keys: Vec<PrivateKeyDer> = rsa_private_keys(&mut kr2)
1500 .filter_map(|r| r.ok())
1501 .map(PrivateKeyDer::from)
1502 .collect();
1503 rsa_keys.into_iter().next()
1504 })
1505 .ok_or_else(|| {
1506 webx_core::error::Error::Http(format!("No valid private key in '{}'", key_path))
1507 })?;
1508
1509 let config = ServerConfig::builder()
1510 .with_no_client_auth()
1511 .with_single_cert(certs, key)
1512 .map_err(|e| webx_core::error::Error::Http(format!("TLS config error: {}", e)))?;
1513
1514 Ok(TlsAcceptor::from(std::sync::Arc::new(config)))
1515}
1516
1517#[cfg(test)]
1518mod embedded_attach_tests {
1519 use super::with_embedded_assets;
1520 use webx_spa::{ContentEncoding, EmbeddedAsset, EmbeddedAssets, SpaSource};
1521
1522 const FILES: &[EmbeddedAsset] = &[EmbeddedAsset {
1523 path: "index.html",
1524 bytes: b"hi",
1525 raw_len: 2,
1526 encoding: ContentEncoding::Identity,
1527 content_type: "text/html",
1528 etag: "\"x\"",
1529 }];
1530
1531 #[test]
1532 fn no_table_leaves_disk_source() {
1533 let source = Some(SpaSource::Disk("wwwroot".into()));
1534 let merged = with_embedded_assets(source, None);
1535 assert!(matches!(merged, Some(SpaSource::Disk(ref r)) if r == "wwwroot"));
1536 }
1537
1538 #[test]
1539 fn table_layers_under_use_spa_overlay() {
1540 let assets = EmbeddedAssets::new("../wwwroot", FILES);
1541 let merged = with_embedded_assets(Some(SpaSource::Disk("wwwroot".into())), Some(assets));
1542 match merged {
1543 Some(SpaSource::Embedded {
1544 overlay_root,
1545 assets,
1546 }) => {
1547 assert_eq!(overlay_root, "wwwroot");
1548 assert_eq!(assets.root(), "../wwwroot");
1549 }
1550 other => panic!("expected embedded, got {other:?}"),
1551 }
1552 }
1553
1554 #[test]
1555 fn table_alone_uses_build_root_as_overlay() {
1556 let assets = EmbeddedAssets::new("../wwwroot", FILES);
1557 let merged = with_embedded_assets(None, Some(assets));
1558 match merged {
1559 Some(SpaSource::Embedded { overlay_root, .. }) => {
1560 assert_eq!(overlay_root, "../wwwroot");
1561 }
1562 other => panic!("expected embedded, got {other:?}"),
1563 }
1564 }
1565
1566 #[test]
1567 fn host_crate_registers_no_asset_table() {
1568 assert!(EmbeddedAssets::registered().is_none());
1569 }
1570}