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::middleware::IMiddleware;
18use webx_core::mode::AppMode;
19use webx_core::routing::{HttpMethod, IEndpoint, IRouter};
20
21use crate::auth_jwt::{init_jwt_secret, jwt_middleware, JwtAuth};
22use crate::authz::{build_resource_policy_from_routes, collect_authorizers};
23use crate::context::{BodyLimits, HttpContext, RespBody};
24use crate::cors::{CorsConfig, CorsMiddleware};
25use crate::endpoint::{StaticHtmlEndpoint, StaticJsonEndpoint, StubEndpoint};
26use crate::health::{HealthCheckEndpoint, HealthCheckRegistry, HealthStatus};
27use crate::memory_cache::MemoryCache;
28use crate::metrics::{HttpMetrics, MetricsEndpoint, MetricsMiddleware};
29use crate::pipeline::{HandlerFn, MiddlewarePipeline};
30use crate::problem_response::{build_problem, write_problem_response};
31use crate::rate_limit::RateLimitMiddleware;
32use crate::router::Router;
33use jsonwebtoken::{DecodingKey, Validation};
34use webx_core::route::scan::RouteEntry;
35use webx_core::DispatchRuntime;
36use webx_openapi::{generate_openapi_spec, APIUI_HTML};
37use webx_spa::{EmbeddedAssets, SpaMiddleware, SpaSource};
38
39use rustls::pki_types::{CertificateDer, PrivateKeyDer};
40use rustls::ServerConfig;
41use rustls_pemfile::{certs, pkcs8_private_keys, rsa_private_keys};
42use std::sync::Arc;
43use std::time::Instant;
44use tokio::task::JoinSet;
45use tokio_rustls::TlsAcceptor;
46
47fn is_weak_jwt_secret(secret: &str) -> bool {
49 secret.is_empty()
50 || secret.len() < 32
51 || secret.contains("change-in-production")
52 || secret.contains("please-change")
53 || secret.contains("demo-secret")
54 || secret.contains("dev-secret")
55}
56
57fn log_spa_source(spa: &SpaMiddleware) {
62 tracing::info!("[Host] SPA root: {}", spa.resolved_root().display());
63 let Some(assets) = spa.embedded() else {
64 return;
65 };
66 tracing::info!(
67 "[Host] Embedded assets: {} files, {:.1} MiB embedded ({:.1} MiB uncompressed), built from `{}`",
68 assets.len(),
69 assets.total_bytes() as f64 / (1024.0 * 1024.0),
70 assets.total_raw_bytes() as f64 / (1024.0 * 1024.0),
71 assets.root(),
72 );
73 let shadowed = assets.shadowed_by(spa.resolved_root());
74 if !shadowed.is_empty() {
75 tracing::warn!(
76 "[Host] {} embedded file(s) are overridden on disk: {}. \
77 An override older than the current build stays in effect until it is \
78 updated or deleted.",
79 shadowed.len(),
80 shadowed.join(", "),
81 );
82 }
83}
84
85fn registered_embedded_assets() -> Option<EmbeddedAssets> {
94 let mut found = inventory::iter::<EmbeddedAssets>();
95 match (found.next(), found.next()) {
96 (Some(assets), None) => Some(*assets),
97 (None, _) => None,
98 (Some(_), Some(_)) => panic!(
99 "more than one compiled-in asset table is registered; \
100 use `#[webx::main(embed)]` (or `#[webx::embed_assets]`) \
101 exactly once per binary"
102 ),
103 }
104}
105
106fn with_embedded_assets(
112 source: Option<SpaSource>,
113 assets: Option<EmbeddedAssets>,
114) -> Option<SpaSource> {
115 let Some(assets) = assets else {
116 return source;
117 };
118 let overlay_root = match source.as_ref() {
119 Some(existing) => existing.disk_root().to_string(),
120 None => assets.root().to_string(),
121 };
122 Some(SpaSource::Embedded {
123 assets,
124 overlay_root,
125 })
126}
127
128pub struct Host {
129 #[allow(dead_code)]
130 provider: Arc<ServiceProvider>,
131 dispatch_runtime: Arc<DispatchRuntime>,
132 pub options: AppOptions,
133 body_limits: BodyLimits,
135 pipeline: Arc<MiddlewarePipeline>,
136 #[allow(dead_code)]
138 router: Arc<Router>,
139 router_handler: HandlerFn,
141 mode: AppMode,
142 #[allow(dead_code)]
144 spa_source: Option<SpaSource>,
145 shutdown: Arc<tokio::sync::Notify>,
146 hosted_services: Vec<Arc<dyn IHostedService>>,
149}
150
151#[allow(clippy::type_complexity)]
152pub struct HostBuilder {
153 service_configs: Vec<Box<dyn FnOnce(ServiceCollection) -> ServiceCollection + Send>>,
154 mode: AppMode,
155 spa_source: Option<SpaSource>,
156 spa_disabled: bool,
157 options_modifiers: Vec<Box<dyn FnOnce(&mut AppOptions) + Send>>,
158 cors_config: Option<CorsConfig>,
159 auth_enabled: bool,
160 resource_auth_enabled: bool,
161 health_registry: Arc<HealthCheckRegistry>,
162}
163
164#[allow(clippy::type_complexity)]
165pub struct HostAppBuilder {
166 options_modifiers: Vec<Box<dyn FnOnce(&mut AppOptions) + Send>>,
167}
168
169impl HostAppBuilder {
170 fn new() -> Self {
171 Self {
172 options_modifiers: Vec::new(),
173 }
174 }
175
176 #[allow(non_snake_case)]
177 pub fn useOptions<F>(&mut self, f: F)
178 where
179 F: FnOnce(&mut AppOptions) + Send + 'static,
180 {
181 self.options_modifiers.push(Box::new(f));
182 }
183}
184
185impl HostBuilder {
186 pub fn new() -> Self {
187 Self {
188 service_configs: Vec::new(),
189 mode: AppMode::from_env(),
192 spa_source: None,
193 spa_disabled: false,
194 options_modifiers: Vec::new(),
195 cors_config: None,
196 auth_enabled: false,
197 resource_auth_enabled: false,
198 health_registry: HealthCheckRegistry::new(),
199 }
200 }
201
202 pub fn register<F>(mut self, f: F) -> Self
203 where
204 F: FnOnce(ServiceCollection) -> ServiceCollection + Send + 'static,
205 {
206 self.service_configs.push(Box::new(f));
207 self
208 }
209
210 pub fn configure<F>(mut self, f: F) -> Self
211 where
212 F: FnOnce(&mut HostAppBuilder) + Send + 'static,
213 {
214 let mut builder = HostAppBuilder::new();
215 f(&mut builder);
216 self.options_modifiers
217 .append(&mut builder.options_modifiers);
218 self
219 }
220
221 pub fn mode(mut self, mode: AppMode) -> Self {
222 self.mode = mode;
223 self
224 }
225
226 pub fn add_options<T>(self, section: &str) -> Self
238 where
239 T: for<'de> serde::Deserialize<'de> + Default + Send + Sync + 'static,
240 {
241 let mode = self.mode;
242 let section = section.to_string();
243 self.register(move |svc| {
244 let appsettings = config::load_appsettings(mode).unwrap_or_default();
245 let bound: T = config::bind_config(&appsettings, §ion);
246 tracing::info!(
247 "[Host] add_options<{}>(\"{}\") registered",
248 std::any::type_name::<T>(),
249 section
250 );
251 svc.instance(Arc::new(bound))
252 })
253 }
254
255 pub fn use_spa(mut self, source: impl Into<SpaSource>) -> Self {
270 self.spa_source = Some(match (self.spa_source.take(), source.into()) {
271 (Some(SpaSource::Embedded { assets, .. }), SpaSource::Disk(root)) => {
274 SpaSource::Embedded {
275 assets,
276 overlay_root: root,
277 }
278 }
279 (_, incoming) => incoming,
280 });
281 self
282 }
283
284 pub fn no_spa(mut self) -> Self {
290 self.spa_disabled = true;
291 self
292 }
293
294 pub fn use_cors(mut self, config: CorsConfig) -> Self {
295 self.cors_config = Some(config);
296 self
297 }
298
299 pub fn add_authentication(mut self) -> Self {
311 self.auth_enabled = true;
312 self
313 }
314
315 pub fn use_resource_authorization(mut self) -> Self {
322 self.resource_auth_enabled = true;
323 self
324 }
325
326 pub fn add_memory_cache(mut self) -> Self {
338 let cache = Arc::new(MemoryCache::new());
339 self.service_configs
340 .push(Box::new(move |svc| svc.instance(cache)));
341 self
342 }
343
344 pub fn add_memory_cache_with(mut self, max_entries: usize) -> Self {
346 let cache = Arc::new(MemoryCache::new().with_max_entries(max_entries));
347 self.service_configs
348 .push(Box::new(move |svc| svc.instance(Arc::clone(&cache))));
349 self
350 }
351
352 pub fn use_middleware<T: IMiddleware + Default + 'static>(mut self) -> Self {
364 self.service_configs.push(Box::new(|svc| {
365 svc.singleton::<dyn IMiddleware>(|_| Arc::new(T::default()) as Arc<dyn IMiddleware>)
366 }));
367 self
368 }
369
370 pub fn use_middleware_with<F>(mut self, factory: F) -> Self
382 where
383 F: Fn() -> Arc<dyn IMiddleware> + Send + Sync + 'static,
384 {
385 self.service_configs.push(Box::new(move |svc| {
386 svc.singleton::<dyn IMiddleware>(move |_| factory())
387 }));
388 self
389 }
390
391 pub fn add_health_check<F>(self, name: impl Into<String>, check: F) -> Self
404 where
405 F: Fn() -> HealthStatus + Send + Sync + 'static,
406 {
407 self.health_registry.register(name, Arc::new(check));
408 self
409 }
410
411 pub fn build(self) -> Host {
412 let env_filter = std::env::var("RUST_LOG").unwrap_or_else(|_| "info".into());
415 if self.mode == AppMode::Development {
416 let _ = tracing_subscriber::fmt()
417 .with_env_filter(env_filter)
418 .without_time()
419 .with_target(false)
420 .try_init();
421 } else {
422 let _ = tracing_subscriber::fmt()
423 .json()
424 .with_env_filter(env_filter)
425 .try_init();
426 }
427
428 let mut svc = ServiceCollection::from_injected();
429 for cfg in self.service_configs {
430 svc = cfg(svc);
431 }
432
433 let health_registry = Arc::clone(&self.health_registry);
435 svc = svc.instance(health_registry);
436
437 let provider = svc.build().unwrap_or_else(|e| {
438 panic!(
439 "Failed to build ServiceProvider: {}. Check your DI registrations.",
440 e
441 )
442 });
443
444 let handler_cache = Arc::new(webx_core::route::scan::HandlerCache::build());
446 let dispatch_runtime = Arc::new(DispatchRuntime::new(Arc::clone(&provider), handler_cache));
447
448 let appsettings =
449 config::load_appsettings(self.mode).unwrap_or_else(|| serde_json::json!({}));
450 let mut options: AppOptions = config::bind_root(&appsettings);
451 for modifier in self.options_modifiers {
452 modifier(&mut options);
453 }
454
455 let body_limits = BodyLimits::from_options(&options);
457 body_limits.apply_spool_root();
458 tracing::info!(
459 "[Host] Uploads: max request {} bytes, max file {} bytes, spool >{} bytes to {}",
460 body_limits.multipart_body,
461 body_limits.file,
462 body_limits.memory_threshold,
463 body_limits.spool_dir.display()
464 );
465
466 let http_metrics = HttpMetrics::new();
467
468 let mut pipeline = MiddlewarePipeline::new();
469
470 pipeline.add_middleware(Arc::new(
473 crate::security_headers::SecurityHeadersMiddleware::new(),
474 ));
475 pipeline.add_middleware(Arc::new(crate::request_id::RequestIdMiddleware::new()));
476
477 if options.rate_limit.enabled {
478 tracing::info!(
479 "[Host] RateLimit enabled: {} req/s, burst {}, max_ips {}",
480 options.rate_limit.requests_per_second,
481 options.rate_limit.burst_size,
482 options.rate_limit.max_tracked_ips
483 );
484 pipeline.add_middleware(Arc::new(RateLimitMiddleware::from_config(
485 &options.rate_limit,
486 )));
487 }
488
489 if options.metrics.enabled {
490 tracing::info!("[Host] Metrics enabled at GET /metrics");
491 pipeline.add_middleware(Arc::new(MetricsMiddleware::new(Arc::clone(&http_metrics))));
492 }
493
494 let middlewares: Vec<Arc<dyn IMiddleware>> = provider.get_all::<dyn IMiddleware>();
495 for mw in middlewares {
496 pipeline.add_middleware(mw);
497 }
498
499 let cors = self.cors_config.unwrap_or_else(|| {
500 let cs = &options.cors;
501 CorsConfig {
502 origins: cs.origins.clone(),
503 methods: cs.methods.clone(),
504 headers: cs.headers.clone(),
505 expose_headers: cs.expose_headers.clone(),
506 allow_credentials: cs.allow_credentials,
507 max_age: cs.max_age,
508 }
509 });
510 pipeline.add_middleware(Arc::new(CorsMiddleware::new(cors)));
511
512 if self.auth_enabled {
513 let secret = options.jwt.secret.clone();
514 if !secret.is_empty() {
515 let jwt_auth = Arc::new(JwtAuth::new(
516 DecodingKey::from_secret(secret.as_bytes()),
517 Validation::default(),
518 ));
519 pipeline.add_middleware(Arc::new(jwt_middleware(jwt_auth)));
520 init_jwt_secret(&secret);
521 } else {
522 tracing::warn!(
523 "add_authentication() enabled but no JWT secret configured. Set Jwt.Secret in appsettings.json, JWT_SECRET, or APP__Jwt__Secret env var."
524 );
525 }
526 }
527
528 let spa_source = if self.spa_disabled {
536 None
537 } else {
538 let base = self.spa_source.clone().or_else(|| {
539 let candidate = webx_core::paths::app_base().join("wwwroot");
540 if candidate.is_dir() {
541 tracing::info!("[Host] Auto-detected SPA root: {}", candidate.display());
542 Some(SpaSource::Disk(candidate.to_string_lossy().into_owned()))
543 } else {
544 None
545 }
546 });
547 with_embedded_assets(base, registered_embedded_assets())
548 };
549 if let Some(ref source) = spa_source {
550 let spa = SpaMiddleware::from_source(source.clone());
551 log_spa_source(&spa);
552 pipeline.add_middleware(Arc::new(spa));
553 }
554
555 let pipeline = Arc::new(pipeline);
560
561 if self.mode == AppMode::Production
563 && self.auth_enabled
564 && is_weak_jwt_secret(&options.jwt.secret)
565 {
566 panic!(
567 "Production requires a strong JWT secret. Set APP__Jwt__Secret or JWT_SECRET env var (min 32 chars, not a placeholder)."
568 );
569 }
570
571 if self.mode == AppMode::Production && options.cors.origins.iter().any(|o| o == "*") {
573 panic!(
574 "Production forbids CORS origin '*'. Set explicit origins via Cors.Origins or APP__Cors__Origins."
575 );
576 }
577
578 let mut router = Router::new();
579 let mut route_count = 0usize;
580
581 crate::diagnostics::assert_route_configuration_valid();
582
583 let mut dispatch_map: std::collections::HashMap<
585 &'static str,
586 webx_core::route::scan::RouteDispatchFn,
587 > = std::collections::HashMap::new();
588
589 for dispatch in inventory::iter::<webx_core::route::scan::RouteDispatch> {
590 dispatch_map.insert(dispatch.handler_type, dispatch.dispatch);
591 }
592
593 let resource_policy: Option<Arc<crate::authz::ResourceAuthorization>> =
594 if self.resource_auth_enabled {
595 Some(Arc::new(build_resource_policy_from_routes()))
596 } else {
597 None
598 };
599
600 for entry in inventory::iter::<RouteEntry> {
601 route_count += 1;
602 let stub = Arc::new(StubEndpoint {
603 method: entry.method.as_str(),
604 path: entry.path,
605 handler_type: entry.handler_type,
606 dispatch_fn: dispatch_map.get(entry.handler_type).copied(),
607 auth_required_role: entry.required_role,
608 auth_required_permission: entry.required_permission,
609 resource_policy: resource_policy.clone(),
610 authorizers: collect_authorizers(provider.as_ref()),
611 });
612 router.register(entry.method, entry.path, stub);
613 }
614
615 let openapi_spec = generate_openapi_spec("Rust WebX API", env!("CARGO_PKG_VERSION"));
616 if self.mode == AppMode::Development {
617 let openapi_bytes = serde_json::to_vec(&openapi_spec).unwrap_or_default();
618 router.register(
619 HttpMethod::Get,
620 "/api/openapi.json",
621 Arc::new(StaticJsonEndpoint::new(openapi_bytes)),
622 );
623 router.register(
624 HttpMethod::Get,
625 "/api/openapi.html",
626 Arc::new(StaticHtmlEndpoint { body: APIUI_HTML }),
627 );
628 }
629
630 let health_registry = Arc::clone(&self.health_registry);
633 let health_endpoint: Arc<dyn IEndpoint> =
634 Arc::new(HealthCheckEndpoint::new(Arc::clone(&health_registry)));
635 router.register(HttpMethod::Get, "/health", Arc::clone(&health_endpoint));
636 router.register(HttpMethod::Get, "/healthz", health_endpoint);
637
638 let live_body =
640 serde_json::to_vec(&serde_json::json!({"status":"pass"})).unwrap_or_default();
641 router.register(
642 HttpMethod::Get,
643 "/health/live",
644 Arc::new(StaticJsonEndpoint {
645 body: live_body,
646 content_type: "application/health+json",
647 }),
648 );
649
650 router.register(
652 HttpMethod::Get,
653 "/health/ready",
654 Arc::new(HealthCheckEndpoint::new(health_registry)),
655 );
656
657 if options.metrics.enabled {
658 router.register(
659 HttpMethod::Get,
660 "/metrics",
661 Arc::new(MetricsEndpoint::new(http_metrics)),
662 );
663 }
664
665 if self.mode == AppMode::Development {
666 let version = env!("CARGO_PKG_VERSION");
667 tracing::info!("");
668 tracing::info!(" ----------------------------------------------------------------");
669 tracing::info!(" Rust WebX Framework v{}", version);
670 tracing::info!(" ----------------------------------------------------------------");
671 tracing::info!(" App: {}", options.app.name);
672 tracing::info!(" CORS: enabled");
673 if let Some(source) = spa_source.as_ref() {
674 tracing::info!(" SPA Root: {}", source.disk_root());
675 if let Some(assets) = source.embedded() {
676 tracing::info!(
677 " Embedded: {} files, {:.1} MiB from {:.1} MiB (built from {})",
678 assets.len(),
679 assets.total_bytes() as f64 / (1024.0 * 1024.0),
680 assets.total_raw_bytes() as f64 / (1024.0 * 1024.0),
681 assets.root(),
682 );
683 }
684 }
685 if route_count > 0 {
686 tracing::info!(" Routes: {} registered", route_count);
687 }
688 crate::diagnostics::log_startup_diagnostics();
689 let banner_urls = if options.app.urls.is_empty() {
690 vec!["http://localhost:5000".to_string()]
691 } else {
692 options
693 .app
694 .urls
695 .iter()
696 .map(|u| u.replace("0.0.0.0", "localhost"))
697 .collect::<Vec<_>>()
698 };
699 for url in &banner_urls {
700 tracing::info!(" OpenAPI: {}/api/openapi.html", url);
701 tracing::info!(" OpenAPI: {}/api/openapi.json", url);
702 }
703 tracing::info!(" ----------------------------------------------------------------");
704 tracing::info!("");
705 } else if route_count > 0 {
706 tracing::info!("{} route(s) registered", route_count);
707 }
708
709 let router = Arc::new(router);
710 let router_handler = make_router_handler(Arc::clone(&router));
711
712 let hosted_services: Vec<Arc<dyn IHostedService>> =
715 provider.get_all::<dyn IHostedService>();
716
717 Host {
718 provider,
719 dispatch_runtime,
720 options,
721 body_limits,
722 pipeline,
723 router,
724 router_handler,
725 mode: self.mode,
726 spa_source,
727 shutdown: Arc::new(tokio::sync::Notify::new()),
728 hosted_services,
729 }
730 }
731}
732
733pub struct ServerHandle {
735 shutdown: Arc<tokio::sync::Notify>,
736}
737
738impl ServerHandle {
739 pub fn shutdown(self) {
742 self.shutdown.notify_waiters();
743 }
744}
745
746pub struct Server {
758 host: Host,
759}
760
761impl Server {
762 pub async fn run(self) -> Result<()> {
764 self.host.run().await
765 }
766
767 pub async fn run_at(self, addr: &str) -> Result<()> {
769 self.host.run_at(addr).await
770 }
771
772 pub fn handle(&self) -> ServerHandle {
774 self.host.server_handle()
775 }
776}
777
778impl Default for HostBuilder {
779 fn default() -> Self {
780 Self::new()
781 }
782}
783
784impl Host {
785 pub fn builder() -> HostBuilder {
786 HostBuilder::new()
787 }
788
789 pub fn options(&self) -> &AppOptions {
790 &self.options
791 }
792
793 pub fn provider(&self) -> &Arc<ServiceProvider> {
795 &self.provider
796 }
797
798 pub fn dispatch_runtime(&self) -> &Arc<DispatchRuntime> {
800 &self.dispatch_runtime
801 }
802
803 pub async fn run(&self) -> Result<()> {
817 start_hosted_services(&self.hosted_services, &self.dispatch_runtime).await?;
818
819 let urls = if self.options.app.urls.is_empty() {
820 vec!["http://0.0.0.0:5000".to_string()]
821 } else {
822 self.options.app.urls.clone()
823 };
824
825 let mut http_addrs: Vec<String> = Vec::new();
826 let mut https_addrs: Vec<String> = Vec::new();
827
828 for url in &urls {
829 let (scheme, addr) = parse_url(url)?;
830 match scheme {
831 "http" => http_addrs.push(addr),
832 "https" => https_addrs.push(addr),
833 other => {
834 return Err(webx_core::error::Error::Http(format!(
835 "Unsupported URL scheme '{}' in '{}'",
836 other, url
837 )))
838 }
839 }
840 }
841
842 let acceptor = if !https_addrs.is_empty() {
843 let tls = &self.options.tls;
844 if tls.cert_path.is_empty() || tls.key_path.is_empty() {
845 return Err(webx_core::error::Error::Http(
846 "HTTPS URLs require Tls.CertPath and Tls.KeyPath".into(),
847 ));
848 }
849 Some(build_tls_acceptor(&tls.cert_path, &tls.key_path)?)
850 } else {
851 None
852 };
853
854 if self.mode == AppMode::Development {
856 tracing::info!("");
857 for url in &urls {
858 let display_url = url.replace("0.0.0.0", "localhost");
859 tracing::info!(" Listening on {}", display_url);
860 }
861 } else {
862 tracing::info!("Listening on {} url(s)", urls.len());
863 }
864
865 let notify = Arc::clone(&self.shutdown);
866 install_shutdown_handler(Arc::clone(¬ify));
867 spawn_async_shutdown_listener(Arc::clone(¬ify));
868
869 let mut handles = Vec::new();
870 let pipeline = Arc::clone(&self.pipeline);
871 let router_handler = self.router_handler.clone();
872 let dispatch_runtime = Arc::clone(&self.dispatch_runtime);
873 let mode = self.mode;
874 let body_limits = self.body_limits.clone();
875 let max_connections = self.options.app.max_connections;
876
877 for addr in &http_addrs {
878 let addr = addr.clone();
879 let n = std::sync::Arc::clone(¬ify);
880 let p = Arc::clone(&pipeline);
881 let rh = router_handler.clone();
882 let dr = Arc::clone(&dispatch_runtime);
883 handles.push(tokio::spawn(serve_http(
884 addr,
885 n,
886 p,
887 rh,
888 dr,
889 mode,
890 body_limits.clone(),
891 max_connections,
892 )));
893 }
894
895 if let Some(ref tls_acceptor) = acceptor {
896 for addr in &https_addrs {
897 let addr = addr.clone();
898 let n = std::sync::Arc::clone(¬ify);
899 let p = Arc::clone(&pipeline);
900 let rh = router_handler.clone();
901 let dr = Arc::clone(&dispatch_runtime);
902 let a = tls_acceptor.clone();
903 handles.push(tokio::spawn(serve_https(
904 addr,
905 a,
906 n,
907 p,
908 rh,
909 dr,
910 mode,
911 body_limits.clone(),
912 max_connections,
913 )));
914 }
915 }
916
917 for h in handles {
919 let _ = h.await;
920 }
921
922 stop_hosted_services(&self.hosted_services).await;
923
924 Ok(())
925 }
926
927 pub async fn run_at(&self, addr: &str) -> Result<()> {
929 start_hosted_services(&self.hosted_services, &self.dispatch_runtime).await?;
930
931 let notify = Arc::clone(&self.shutdown);
932 install_shutdown_handler(Arc::clone(¬ify));
933 spawn_async_shutdown_listener(Arc::clone(¬ify));
934 serve_http(
935 addr.to_string(),
936 notify,
937 Arc::clone(&self.pipeline),
938 self.router_handler.clone(),
939 Arc::clone(&self.dispatch_runtime),
940 self.mode,
941 self.body_limits.clone(),
942 self.options.app.max_connections,
943 )
944 .await;
945
946 stop_hosted_services(&self.hosted_services).await;
947 Ok(())
948 }
949
950 pub fn server_handle(&self) -> ServerHandle {
961 ServerHandle {
962 shutdown: Arc::clone(&self.shutdown),
963 }
964 }
965
966 pub fn into_server(self) -> Server {
971 Server { host: self }
972 }
973}
974
975#[async_trait::async_trait]
976impl IHost for Host {
977 async fn run(&self, addr: &str) -> Result<()> {
978 self.run_at(addr).await
979 }
980 async fn stop(&self) -> Result<()> {
981 tracing::info!("Stop requested.");
982 self.shutdown.notify_waiters();
983 Ok(())
984 }
985}
986
987fn make_router_handler(router: Arc<Router>) -> HandlerFn {
988 Arc::new(move |ctx: &mut dyn IHttpContext| {
989 let router = Arc::clone(&router);
990 Box::pin(async move {
991 match router.match_route(ctx).await? {
992 Some((endpoint, params, pattern)) => {
993 drop(router);
994 for (key, value) in params {
995 ctx.request_mut().route_params_mut().insert(key, value);
996 }
997 *ctx.request_mut().route_pattern_mut() = Some(pattern);
998 endpoint.handle(ctx).await
999 }
1000 None => {
1001 let path_exists = router.path_exists(ctx.request().path());
1002 drop(router);
1003 if !ctx.response().has_body() {
1006 if path_exists {
1007 write_error_response(ctx, 405, "Method Not Allowed").await;
1008 } else {
1009 write_error_response(ctx, 404, "Not Found").await;
1010 }
1011 }
1012 Ok(())
1013 }
1014 }
1015 })
1016 })
1017}
1018
1019async fn handle_request(
1020 req: Request<Incoming>,
1021 pipeline: Arc<MiddlewarePipeline>,
1022 router_handler: HandlerFn,
1023 dispatch_runtime: Arc<DispatchRuntime>,
1024 body_limits: BodyLimits,
1025) -> std::result::Result<hyper::Response<RespBody>, std::convert::Infallible> {
1026 let mut ctx = HttpContext::new(req, body_limits);
1027
1028 if ctx.response().status() < 400 {
1031 let result = dispatch_runtime
1032 .run(async { pipeline.execute(&mut ctx, router_handler).await })
1033 .await;
1034 if let Err(e) = result {
1035 let status = e.status_code();
1036 write_error_response(&mut ctx, status, &e.to_string()).await;
1037 }
1038 }
1039
1040 Ok(ctx.into_response().await)
1041}
1042
1043async fn write_error_response(ctx: &mut dyn IHttpContext, status: u16, message: &str) {
1044 write_problem_response(ctx, build_problem(status, message)).await;
1045}
1046
1047fn install_shutdown_handler(shutdown: Arc<tokio::sync::Notify>) {
1057 use std::sync::atomic::{AtomicUsize, Ordering};
1058
1059 static CTRL_COUNT: AtomicUsize = AtomicUsize::new(0);
1060
1061 if let Err(e) = ctrlc::set_handler(move || {
1062 let n = CTRL_COUNT.fetch_add(1, Ordering::SeqCst);
1063 if n == 0 {
1064 tracing::info!("Shutdown signal received, draining connections...");
1065 shutdown.notify_waiters();
1066 } else {
1067 tracing::warn!("Second interrupt — force exiting.");
1068 std::process::exit(130);
1069 }
1070 }) {
1071 tracing::warn!("Could not install Ctrl+C handler: {}", e);
1072 }
1073}
1074
1075fn spawn_async_shutdown_listener(shutdown: Arc<tokio::sync::Notify>) {
1077 tokio::spawn(async move {
1078 wait_for_shutdown_signal().await;
1079 tracing::info!("Async shutdown signal received, draining connections...");
1080 shutdown.notify_waiters();
1081 });
1082}
1083
1084async fn wait_for_shutdown_signal() {
1085 #[cfg(unix)]
1086 {
1087 use tokio::signal::unix::{signal, SignalKind};
1088 let mut sigterm = signal(SignalKind::terminate()).expect("install SIGTERM handler");
1089 tokio::select! {
1090 res = tokio::signal::ctrl_c() => {
1091 if let Err(e) = res {
1092 tracing::warn!("Ctrl+C listener error: {}", e);
1093 }
1094 }
1095 _ = sigterm.recv() => {}
1096 }
1097 }
1098 #[cfg(not(unix))]
1099 {
1100 if let Err(e) = tokio::signal::ctrl_c().await {
1101 tracing::warn!("Ctrl+C listener error: {}", e);
1102 }
1103 }
1104}
1105
1106async fn start_hosted_services(
1107 services: &[Arc<dyn IHostedService>],
1108 dispatch_runtime: &DispatchRuntime,
1109) -> Result<()> {
1110 if services.is_empty() {
1111 return Ok(());
1112 }
1113 tracing::info!("Starting {} hosted service(s)...", services.len());
1114 dispatch_runtime
1115 .run(async {
1116 for svc in services {
1117 svc.start().await?;
1118 }
1119 Ok::<(), webx_core::error::Error>(())
1120 })
1121 .await?;
1122 tracing::info!("All hosted services started.");
1123 Ok(())
1124}
1125
1126async fn stop_hosted_services(services: &[Arc<dyn IHostedService>]) {
1127 if services.is_empty() {
1128 return;
1129 }
1130 tracing::info!("Stopping {} hosted service(s)...", services.len());
1131 for svc in services {
1132 if let Err(e) = svc.stop().await {
1133 tracing::warn!("Hosted service stop error: {}", e);
1134 }
1135 }
1136 tracing::info!("All hosted services stopped.");
1137}
1138
1139async fn drain_connections(join_set: &mut JoinSet<()>, label: &str, mode: AppMode) {
1140 let timeout_secs = if mode == AppMode::Development { 5 } else { 30 };
1141 let drain_future = async {
1142 while let Some(result) = join_set.join_next().await {
1143 if let Err(e) = result {
1144 tracing::error!("Connection task panicked: {:?}", e);
1145 }
1146 }
1147 };
1148
1149 match tokio::time::timeout(std::time::Duration::from_secs(timeout_secs), drain_future).await {
1150 Ok(_) => tracing::info!("{}: drained.", label),
1151 Err(_) => {
1152 tracing::warn!("{}: drain timeout, aborting connections.", label);
1153 join_set.abort_all();
1154 while join_set.join_next().await.is_some() {}
1155 }
1156 }
1157}
1158
1159fn parse_url(url: &str) -> Result<(&str, String)> {
1166 if let Some(rest) = url.strip_prefix("https://") {
1167 Ok(("https", rest.to_string()))
1168 } else if let Some(rest) = url.strip_prefix("http://") {
1169 Ok(("http", rest.to_string()))
1170 } else {
1171 Err(webx_core::error::Error::Http(format!(
1172 "Invalid URL '{}'. Use http://host:port or https://host:port",
1173 url
1174 )))
1175 }
1176}
1177
1178#[allow(clippy::too_many_arguments)]
1180async fn serve_http(
1181 addr: String,
1182 shutdown: std::sync::Arc<tokio::sync::Notify>,
1183 pipeline: Arc<MiddlewarePipeline>,
1184 router_handler: HandlerFn,
1185 dispatch_runtime: Arc<DispatchRuntime>,
1186 mode: AppMode,
1187 body_limits: BodyLimits,
1188 max_connections: usize,
1189) {
1190 let listener = match tokio::net::TcpListener::bind(&addr).await {
1191 Ok(l) => l,
1192 Err(e) => {
1193 tracing::error!("Failed to bind HTTP on {}: {}", addr, e);
1194 return;
1195 }
1196 };
1197
1198 let sem = Arc::new(tokio::sync::Semaphore::new(max_connections));
1199 let mut join_set = JoinSet::new();
1200
1201 loop {
1202 tokio::select! {
1203 accept_result = listener.accept() => {
1204 let (stream, _) = match accept_result {
1205 Ok(v) => v,
1206 Err(e) => {
1207 tracing::error!("Accept error (will retry): {}", e);
1208 tokio::time::sleep(std::time::Duration::from_millis(100)).await;
1209 continue;
1210 }
1211 };
1212
1213 while join_set.try_join_next().is_some() {}
1214
1215 let permit = match sem.clone().acquire_owned().await {
1216 Ok(p) => p,
1217 Err(_) => continue,
1218 };
1219
1220 let io = TokioIo::new(stream);
1221 let pipeline = Arc::clone(&pipeline);
1222 let router_handler = router_handler.clone();
1223 let dispatch_runtime = Arc::clone(&dispatch_runtime);
1224 let body_limits = body_limits.clone();
1225
1226 join_set.spawn(async move {
1227 let _permit = permit;
1228 let svc_fn = service_fn(move |req: Request<Incoming>| {
1229 let pipeline = Arc::clone(&pipeline);
1230 let router_handler = router_handler.clone();
1231 let dispatch_runtime = Arc::clone(&dispatch_runtime);
1232 let body_limits = body_limits.clone();
1233 let mode = mode;
1234 async move {
1235 let start = Instant::now();
1236 let method = req.method().to_string();
1237 let path = req.uri().path().to_string();
1238 let result = handle_request(
1239 req,
1240 pipeline,
1241 router_handler,
1242 dispatch_runtime,
1243 body_limits,
1244 )
1245 .await;
1246 let elapsed = start.elapsed();
1247 if mode == AppMode::Development {
1248 let status =
1249 result.as_ref().map(|r| r.status().as_u16()).unwrap_or(500);
1250 tracing::info!(
1251 "[{}] {} -> {} ({:.0}ms)",
1252 method,
1253 path,
1254 status,
1255 elapsed.as_secs_f64() * 1000.0
1256 );
1257 }
1258 result
1259 }
1260 });
1261
1262 if let Err(err) = hyper::server::conn::http1::Builder::new()
1263 .serve_connection(io, svc_fn)
1264 .await
1265 {
1266 tracing::error!("Connection error: {}", err);
1267 }
1268 });
1269 }
1270 _ = shutdown.notified() => {
1271 tracing::info!("HTTP {}: stop accepting, draining...", addr);
1272 break;
1273 }
1274 }
1275 }
1276
1277 drain_connections(&mut join_set, &format!("HTTP {}", addr), mode).await;
1278}
1279
1280#[allow(clippy::too_many_arguments)]
1282async fn serve_https(
1283 addr: String,
1284 acceptor: TlsAcceptor,
1285 shutdown: std::sync::Arc<tokio::sync::Notify>,
1286 pipeline: Arc<MiddlewarePipeline>,
1287 router_handler: HandlerFn,
1288 dispatch_runtime: Arc<DispatchRuntime>,
1289 mode: AppMode,
1290 body_limits: BodyLimits,
1291 max_connections: usize,
1292) {
1293 let listener = match tokio::net::TcpListener::bind(&addr).await {
1294 Ok(l) => l,
1295 Err(e) => {
1296 tracing::error!("Failed to bind HTTPS on {}: {}", addr, e);
1297 return;
1298 }
1299 };
1300
1301 let sem = Arc::new(tokio::sync::Semaphore::new(max_connections));
1302 let mut join_set = JoinSet::new();
1303
1304 loop {
1305 tokio::select! {
1306 accept_result = listener.accept() => {
1307 let (stream, _) = match accept_result {
1308 Ok(v) => v,
1309 Err(e) => {
1310 tracing::error!("Accept error (will retry): {}", e);
1311 tokio::time::sleep(std::time::Duration::from_millis(100)).await;
1312 continue;
1313 }
1314 };
1315
1316 while join_set.try_join_next().is_some() {}
1317
1318 let permit = match sem.clone().acquire_owned().await {
1319 Ok(p) => p,
1320 Err(_) => continue,
1321 };
1322
1323 let acceptor = acceptor.clone();
1324 let pipeline = Arc::clone(&pipeline);
1325 let router_handler = router_handler.clone();
1326 let dispatch_runtime = Arc::clone(&dispatch_runtime);
1327 let body_limits = body_limits.clone();
1328
1329 join_set.spawn(async move {
1330 let _permit = permit;
1331 match acceptor.accept(stream).await {
1332 Ok(tls_stream) => {
1333 let io = TokioIo::new(tls_stream);
1334 let svc_fn = service_fn(move |req: Request<Incoming>| {
1335 let pipeline = Arc::clone(&pipeline);
1336 let router_handler = router_handler.clone();
1337 let dispatch_runtime = Arc::clone(&dispatch_runtime);
1338 let body_limits = body_limits.clone();
1339 let mode = mode;
1340 async move {
1341 let start = Instant::now();
1342 let method = req.method().to_string();
1343 let path = req.uri().path().to_string();
1344 let result = handle_request(
1345 req,
1346 pipeline,
1347 router_handler,
1348 dispatch_runtime,
1349 body_limits,
1350 )
1351 .await;
1352 let elapsed = start.elapsed();
1353 if mode == AppMode::Development {
1354 let status =
1355 result.as_ref().map(|r| r.status().as_u16()).unwrap_or(500);
1356 tracing::info!(
1357 "[{}] {} -> {} ({:.0}ms)",
1358 method,
1359 path,
1360 status,
1361 elapsed.as_secs_f64() * 1000.0
1362 );
1363 }
1364 result
1365 }
1366 });
1367
1368 if let Err(err) = hyper::server::conn::http1::Builder::new()
1369 .serve_connection(io, svc_fn)
1370 .await
1371 {
1372 tracing::error!("TLS connection error: {}", err);
1373 }
1374 }
1375 Err(e) => {
1376 tracing::error!("TLS handshake error: {}", e);
1377 }
1378 }
1379 });
1380 }
1381 _ = shutdown.notified() => {
1382 tracing::info!("HTTPS {}: stop accepting, draining...", addr);
1383 break;
1384 }
1385 }
1386 }
1387
1388 drain_connections(&mut join_set, &format!("HTTPS {}", addr), mode).await;
1389}
1390
1391fn build_tls_acceptor(cert_path: &str, key_path: &str) -> Result<TlsAcceptor> {
1397 use std::fs::File;
1398 use std::io::BufReader;
1399
1400 if cert_path.is_empty() || key_path.is_empty() {
1401 return Err(webx_core::error::Error::Http(
1402 "TLS certificate or key path not configured.".into(),
1403 ));
1404 }
1405
1406 let cert_file = File::open(cert_path).map_err(|e| {
1407 webx_core::error::Error::Http(format!("Cannot open cert '{}': {}", cert_path, e))
1408 })?;
1409 let mut cert_reader = BufReader::new(cert_file);
1410 let certs: Vec<CertificateDer> = certs(&mut cert_reader).filter_map(|r| r.ok()).collect();
1411 if certs.is_empty() {
1412 return Err(webx_core::error::Error::Http(format!(
1413 "No valid certs in '{}'",
1414 cert_path
1415 )));
1416 }
1417
1418 let key_file = File::open(key_path).map_err(|e| {
1419 webx_core::error::Error::Http(format!("Cannot open key '{}': {}", key_path, e))
1420 })?;
1421 let mut key_reader = BufReader::new(key_file);
1422 let key = pkcs8_private_keys(&mut key_reader)
1423 .filter_map(|r| r.ok())
1424 .map(PrivateKeyDer::from)
1425 .next()
1426 .or_else(|| {
1427 let key_file2 = File::open(key_path).map(BufReader::new).ok()?;
1428 let mut kr2 = key_file2;
1429 let rsa_keys: Vec<PrivateKeyDer> = rsa_private_keys(&mut kr2)
1430 .filter_map(|r| r.ok())
1431 .map(PrivateKeyDer::from)
1432 .collect();
1433 rsa_keys.into_iter().next()
1434 })
1435 .ok_or_else(|| {
1436 webx_core::error::Error::Http(format!("No valid private key in '{}'", key_path))
1437 })?;
1438
1439 let config = ServerConfig::builder()
1440 .with_no_client_auth()
1441 .with_single_cert(certs, key)
1442 .map_err(|e| webx_core::error::Error::Http(format!("TLS config error: {}", e)))?;
1443
1444 Ok(TlsAcceptor::from(std::sync::Arc::new(config)))
1445}
1446
1447#[cfg(test)]
1448mod embedded_attach_tests {
1449 use super::{registered_embedded_assets, with_embedded_assets};
1450 use webx_spa::{ContentEncoding, EmbeddedAsset, EmbeddedAssets, SpaSource};
1451
1452 const FILES: &[EmbeddedAsset] = &[EmbeddedAsset {
1453 path: "index.html",
1454 bytes: b"hi",
1455 raw_len: 2,
1456 encoding: ContentEncoding::Identity,
1457 content_type: "text/html",
1458 etag: "\"x\"",
1459 }];
1460
1461 #[test]
1462 fn no_table_leaves_disk_source() {
1463 let source = Some(SpaSource::Disk("wwwroot".into()));
1464 let merged = with_embedded_assets(source, None);
1465 assert!(matches!(merged, Some(SpaSource::Disk(ref r)) if r == "wwwroot"));
1466 }
1467
1468 #[test]
1469 fn table_layers_under_use_spa_overlay() {
1470 let assets = EmbeddedAssets::new("../wwwroot", FILES);
1471 let merged = with_embedded_assets(Some(SpaSource::Disk("wwwroot".into())), Some(assets));
1472 match merged {
1473 Some(SpaSource::Embedded {
1474 overlay_root,
1475 assets,
1476 }) => {
1477 assert_eq!(overlay_root, "wwwroot");
1478 assert_eq!(assets.root(), "../wwwroot");
1479 }
1480 other => panic!("expected embedded, got {other:?}"),
1481 }
1482 }
1483
1484 #[test]
1485 fn table_alone_uses_build_root_as_overlay() {
1486 let assets = EmbeddedAssets::new("../wwwroot", FILES);
1487 let merged = with_embedded_assets(None, Some(assets));
1488 match merged {
1489 Some(SpaSource::Embedded { overlay_root, .. }) => {
1490 assert_eq!(overlay_root, "../wwwroot");
1491 }
1492 other => panic!("expected embedded, got {other:?}"),
1493 }
1494 }
1495
1496 #[test]
1497 fn host_crate_registers_no_asset_table() {
1498 assert!(registered_embedded_assets().is_none());
1499 }
1500}