1use crate::manifest::{ErrorRoutes, Manifest, Requires, Route, RouteMode};
33use crate::proxy::cache::{cache_window, compose_key, CacheEntry, CacheState, KeyInputs, ResponseCache};
34use crate::proxy::metrics::Metrics;
35use crate::proxy::session::{SessionResolver, User};
36use crate::proxy::source_gen::Generations;
37use crate::proxy::trace::TraceParent;
38use crate::proxy::worker_client::{RenderResult, WorkerError};
39use crate::proxy::worker_pool::WorkerPool;
40use axum::body::Body;
41use axum::extract::{Request, State};
42use axum::http::{header, HeaderMap, HeaderName, HeaderValue, StatusCode};
43use axum::response::Response;
44use std::collections::BTreeMap;
45use std::path::PathBuf;
46use std::sync::atomic::{AtomicU32, Ordering};
47use std::sync::Arc;
48use std::time::{Duration, Instant};
49use tokio::sync::RwLock;
50use tracing::{debug, warn};
51
52static RENDER_ID: AtomicU32 = AtomicU32::new(1);
55
56const RENDER_DEADLINE_MS: u64 = 2000;
59
60const DEFAULT_NEGATIVE_TTL: u64 = 10;
62
63pub struct AppState {
65 pub manifest: Arc<Manifest>,
66 pub matcher: matchit::Router<usize>,
68 pub pool: Arc<WorkerPool>,
69 pub cdn_base_url: Option<String>,
71 pub fallback_dir: Option<PathBuf>,
73 pub cache: Arc<ResponseCache>,
75 pub generations: Arc<Generations>,
77 pub session: Option<Arc<dyn SessionResolver>>,
80 pub login_url: Option<String>,
83 pub metrics: Arc<Metrics>,
86}
87
88impl AppState {
89 pub fn new(
90 manifest: Arc<Manifest>,
91 pool: Arc<WorkerPool>,
92 cdn_base_url: Option<String>,
93 fallback_dir: Option<PathBuf>,
94 ) -> Self {
95 let matcher = build_matcher(&manifest);
96 Self {
97 manifest,
98 matcher,
99 pool,
100 cdn_base_url,
101 fallback_dir,
102 cache: Arc::new(ResponseCache::new()),
103 generations: Arc::new(Generations::empty()),
104 session: None,
105 login_url: None,
106 metrics: Arc::new(Metrics::new()),
107 }
108 }
109
110 pub fn with_metrics(mut self, metrics: Arc<Metrics>) -> Self {
111 self.metrics = metrics;
112 self
113 }
114
115 pub fn with_cache(mut self, cache: Arc<ResponseCache>) -> Self {
116 self.cache = cache;
117 self
118 }
119
120 pub fn with_generations(mut self, generations: Arc<Generations>) -> Self {
121 self.generations = generations;
122 self
123 }
124
125 pub fn with_session(mut self, session: Arc<dyn SessionResolver>) -> Self {
126 self.session = Some(session);
127 self
128 }
129
130 pub fn with_login_url(mut self, login_url: Option<String>) -> Self {
131 self.login_url = login_url;
132 self
133 }
134}
135
136fn to_matchit_pattern(route: &str) -> String {
150 route
151 .split('/')
152 .map(|seg| match seg.split_at_checked(1) {
153 Some((":", name)) if !name.is_empty() => format!("{{{name}}}"),
154 Some(("*", name)) if !name.is_empty() => format!("{{*{name}}}"),
155 _ => seg.to_string(),
156 })
157 .collect::<Vec<_>>()
158 .join("/")
159}
160
161pub fn build_matcher(manifest: &Manifest) -> matchit::Router<usize> {
163 let mut router = matchit::Router::new();
164 for (i, route) in manifest.routes.iter().enumerate() {
165 let pattern = to_matchit_pattern(&route.route);
166 if let Err(e) = router.insert(&pattern, i) {
167 tracing::warn!("failed to register route '{}': {e}", route.route);
168 }
169 }
170 router
171}
172
173pub type SharedState = Arc<RwLock<AppState>>;
174
175struct SsrCall {
178 path: String,
179 route: Route,
180 build_id: String,
181 params: BTreeMap<String, String>,
182 query: BTreeMap<String, String>,
183 headers: HeaderMap,
184 pool: Arc<WorkerPool>,
185 cache: Arc<ResponseCache>,
186 generations: Arc<Generations>,
187 session: Option<Arc<dyn SessionResolver>>,
188 login_url: Option<String>,
189 metrics: Arc<Metrics>,
190 trace: String,
192}
193
194pub async fn handle(State(state): State<SharedState>, req: Request) -> Response {
198 let path = req.uri().path().to_string();
199 let trace = TraceParent::incoming_or_new(
201 req.headers().get("traceparent").and_then(|v| v.to_str().ok()),
202 );
203
204 enum Plan {
207 NotFound,
208 Static { cdn: Option<String>, fallback: Option<PathBuf> },
209 Ssr(Box<SsrCall>),
210 }
211
212 let (route_label, mode_label, metrics, error_pages, plan) = {
213 let st = state.read().await;
214 let metrics = st.metrics.clone();
215 let error_pages = ErrorPages::from_state(&st);
216 match st.matcher.at(&path) {
217 Err(_) => (
218 "<unmatched>".to_string(),
219 "none",
220 metrics,
221 error_pages,
222 Plan::NotFound,
223 ),
224 Ok(m) => {
225 let idx = *m.value;
226 let params = m
227 .params
228 .iter()
229 .map(|(k, v)| (k.to_string(), v.to_string()))
230 .collect::<BTreeMap<_, _>>();
231 let route = st.manifest.routes[idx].clone();
232 let route_label = route.route.clone();
233 let mode_label = mode_str(&route.mode);
234 debug!(trace = %trace.trace_id, "dispatching {} → mode={:?}", path, route.mode);
235 let plan = match route.mode {
236 RouteMode::Static | RouteMode::Spa => Plan::Static {
239 cdn: st.cdn_base_url.clone(),
240 fallback: st.fallback_dir.clone(),
241 },
242 RouteMode::Ssr => Plan::Ssr(Box::new(SsrCall {
243 path: path.clone(),
244 route,
245 build_id: st.manifest.build_id.clone(),
246 params,
247 query: parse_kv(req.uri().query()),
248 headers: req.headers().clone(),
249 pool: st.pool.clone(),
250 cache: st.cache.clone(),
251 generations: st.generations.clone(),
252 session: st.session.clone(),
253 login_url: st.login_url.clone(),
254 metrics: metrics.clone(),
255 trace: trace.header_value(),
256 })),
257 };
258 (route_label, mode_label, metrics, error_pages, plan)
259 }
260 }
261 };
262
263 let mut resp = match plan {
264 Plan::NotFound => error_pages.not_found().await,
265 Plan::Static { cdn, fallback } => {
266 dispatch_static(&path, cdn, fallback, &error_pages).await
267 }
268 Plan::Ssr(call) => dispatch_ssr(*call).await,
269 };
270
271 if let Ok(v) = HeaderValue::from_str(&trace.header_value()) {
274 resp.headers_mut()
275 .insert(HeaderName::from_static("traceparent"), v);
276 }
277 metrics.record_request(&route_label, mode_label, resp.status().as_u16());
278 resp
279}
280
281pub async fn metrics_handler(State(state): State<SharedState>) -> Response {
285 let (metrics, ready) = {
286 let st = state.read().await;
287 (st.metrics.clone(), st.pool.live_count().await)
288 };
289 Response::builder()
290 .status(StatusCode::OK)
291 .header(header::CONTENT_TYPE, "text/plain; version=0.0.4; charset=utf-8")
292 .body(Body::from(metrics.render(ready)))
293 .unwrap()
294}
295
296fn mode_str(mode: &RouteMode) -> &'static str {
297 match mode {
298 RouteMode::Static => "static",
299 RouteMode::Ssr => "ssr",
300 RouteMode::Spa => "spa",
301 }
302}
303
304async fn dispatch_static(
307 path: &str,
308 cdn: Option<String>,
309 fallback: Option<PathBuf>,
310 error_pages: &ErrorPages,
311) -> Response {
312 if let Some(cdn) = cdn {
314 let target = format!("{cdn}{path}");
315 return Response::builder()
316 .status(StatusCode::FOUND)
317 .header(header::LOCATION, target)
318 .body(Body::empty())
319 .unwrap();
320 }
321
322 if let Some(dir) = fallback {
323 return serve_local(path, &dir, error_pages).await;
324 }
325
326 Response::builder()
328 .status(StatusCode::BAD_GATEWAY)
329 .body(Body::from("no CDN or fallback configured for Mode 1 route"))
330 .unwrap()
331}
332
333async fn serve_local(path: &str, dir: &PathBuf, error_pages: &ErrorPages) -> Response {
338 let relative = path.trim_start_matches('/');
339
340 let candidates: Vec<PathBuf> = if relative.is_empty() {
341 vec![dir.join("index.html")]
342 } else if relative.ends_with('/') {
343 vec![dir.join(relative).join("index.html")]
344 } else {
345 vec![
346 dir.join(format!("{relative}.html")),
347 dir.join(relative).join("index.html"),
348 ]
349 };
350
351 for candidate in &candidates {
352 match tokio::fs::read(candidate).await {
353 Ok(bytes) => {
354 return Response::builder()
355 .status(StatusCode::OK)
356 .header(header::CONTENT_TYPE, "text/html; charset=utf-8")
357 .body(Body::from(bytes))
358 .unwrap();
359 }
360 Err(e) if e.kind() == std::io::ErrorKind::NotFound => continue,
361 Err(e) => {
362 tracing::error!("fallback read error for {}: {e}", candidate.display());
363 return error_pages.internal_error().await;
364 }
365 }
366 }
367
368 error_pages.not_found().await
369}
370
371async fn dispatch_ssr(call: SsrCall) -> Response {
374 let requires_user = call
375 .route
376 .requires
377 .as_ref()
378 .is_some_and(|r| r.contains(&Requires::User));
379
380 let cookie_header = call
382 .headers
383 .get(header::COOKIE)
384 .and_then(|v| v.to_str().ok());
385 let user = call.session.as_ref().and_then(|s| s.resolve(cookie_header));
386
387 if requires_user && user.is_none() {
388 return redirect_to_login(&call.path, &call.query, call.login_url.as_deref());
389 }
390
391 let user_key = if requires_user {
394 Some(user.as_ref().map_or("_anon", |u| u.id.as_str()))
395 } else {
396 None
397 };
398 let vary = collect_vary(&call.route, &call.headers);
399 let gens = collect_generations(&call.route, &call.generations);
400 let key = compose_key(&KeyInputs {
401 build_id: &call.build_id,
402 route_pattern: &call.route.route,
403 params: &call.params,
404 query: &call.query,
405 vary: &vary,
406 source_generations: &gens,
407 user_id: user_key,
408 });
409
410 let cp = &call.route.cache_policy;
411 let ttl = Duration::from_secs(cp.ttl);
412 let swr = Duration::from_secs(cp.swr.unwrap_or(0));
413 let negative_ttl = Duration::from_secs(cp.negative_ttl.unwrap_or(DEFAULT_NEGATIVE_TTL));
414
415 let req_json = build_render_request(&call, user.as_ref());
418
419 if let Some(entry) = call.cache.get(&key) {
421 match entry.state() {
422 CacheState::Fresh => {
423 call.metrics.record_cache(&call.route.route, "fresh");
424 return build_cached_response(&entry, "fresh", false);
425 }
426 CacheState::Stale => {
427 call.metrics.record_cache(&call.route.route, "stale");
429 spawn_refresh(
430 call.pool.clone(),
431 call.cache.clone(),
432 call.metrics.clone(),
433 call.route.route.clone(),
434 req_json.clone(),
435 key.clone(),
436 ttl,
437 swr,
438 negative_ttl,
439 );
440 return build_cached_response(&entry, "stale", false);
441 }
442 CacheState::Expired => {
443 return render_miss(&call, &key, req_json, ttl, swr, negative_ttl, Some(entry)).await;
446 }
447 }
448 }
449
450 render_miss(&call, &key, req_json, ttl, swr, negative_ttl, None).await
451}
452
453async fn render_miss(
457 call: &SsrCall,
458 key: &str,
459 req_json: serde_json::Value,
460 ttl: Duration,
461 swr: Duration,
462 negative_ttl: Duration,
463 fallback: Option<CacheEntry>,
464) -> Response {
465 call.metrics.inflight_inc();
466 let started = Instant::now();
467 let outcome = render_once(&call.pool, &call.route.route, req_json).await;
468 call.metrics
469 .observe_render(&call.route.route, started.elapsed().as_secs_f64());
470 call.metrics.inflight_dec();
471
472 match outcome {
473 Ok(rr) => {
474 let status = 200; let entry = entry_from_render(status, rr, ttl, swr);
476 if let Some((store_ttl, store_swr)) =
477 cache_window(status, ttl, swr, negative_ttl)
478 {
479 let mut to_store = entry.clone();
480 to_store.ttl = store_ttl;
481 to_store.swr = store_swr;
482 call.cache.insert(key.to_string(), to_store);
483 }
484 call.metrics.record_cache(&call.route.route, "miss");
485 build_cached_response(&entry, "miss", false)
486 }
487 Err(e) => {
488 warn!("render failed for {}: {e}", call.route.route);
489 match fallback {
490 Some(entry) => {
492 call.metrics.record_cache(&call.route.route, "stale");
493 build_cached_response(&entry, "stale", true)
494 }
495 None => service_unavailable(ttl),
496 }
497 }
498 }
499}
500
501#[allow(clippy::too_many_arguments)]
504fn spawn_refresh(
505 pool: Arc<WorkerPool>,
506 cache: Arc<ResponseCache>,
507 metrics: Arc<Metrics>,
508 route_pattern: String,
509 req_json: serde_json::Value,
510 key: String,
511 ttl: Duration,
512 swr: Duration,
513 negative_ttl: Duration,
514) {
515 tokio::spawn(async move {
516 metrics.inflight_inc();
517 let started = Instant::now();
518 let outcome = render_once(&pool, &route_pattern, req_json).await;
519 metrics.observe_render(&route_pattern, started.elapsed().as_secs_f64());
520 metrics.inflight_dec();
521 match outcome {
522 Ok(rr) => {
523 let status = 200;
524 if let Some((store_ttl, store_swr)) = cache_window(status, ttl, swr, negative_ttl) {
525 let mut entry = entry_from_render(status, rr, store_ttl, store_swr);
526 entry.ttl = store_ttl;
527 entry.swr = store_swr;
528 cache.insert(key, entry);
529 } else {
530 cache.remove(&key);
531 }
532 }
533 Err(e) => warn!("SWR refresh failed for {route_pattern}: {e}"),
534 }
535 });
536}
537
538async fn render_once(
541 pool: &WorkerPool,
542 route_pattern: &str,
543 req_json: serde_json::Value,
544) -> Result<RenderResult, WorkerError> {
545 let worker = pool.get().await.ok_or(WorkerError::Closed)?;
546 let id = RENDER_ID.fetch_add(1, Ordering::Relaxed);
547 worker.render(id, route_pattern, req_json, RENDER_DEADLINE_MS).await
548}
549
550fn entry_from_render(status: u16, rr: RenderResult, ttl: Duration, swr: Duration) -> CacheEntry {
551 CacheEntry {
552 status,
553 html: rr.html,
554 headers: rr.headers.into_iter().collect(),
555 ttl,
556 swr,
557 stored_at: Instant::now(),
558 }
559}
560
561fn build_render_request(call: &SsrCall, user: Option<&User>) -> serde_json::Value {
564 let url = match call.query.is_empty() {
565 true => call.path.clone(),
566 false => format!("{}?{}", call.path, encode_query(&call.query)),
567 };
568 let headers: BTreeMap<&str, String> = call
569 .headers
570 .iter()
571 .map(|(k, v)| (k.as_str(), v.to_str().unwrap_or("").to_string()))
572 .collect();
573 let cookies = parse_cookies(&call.headers);
574
575 let mut req = serde_json::json!({
576 "url": url,
577 "params": call.params,
578 "query": call.query,
579 "headers": headers,
580 "cookies": cookies,
581 "ctx": { "trace": call.trace },
584 });
585 if let Some(u) = user {
586 req["user"] = serde_json::json!({ "id": u.id, "attrs": u.attrs });
587 }
588 req
589}
590
591fn collect_vary(route: &Route, headers: &HeaderMap) -> BTreeMap<String, String> {
595 let mut out = BTreeMap::new();
596 if let Some(vary) = &route.cache_policy.vary {
597 for name in vary {
598 let value = headers
599 .get(name.as_str())
600 .and_then(|v| v.to_str().ok())
601 .unwrap_or("")
602 .to_string();
603 out.insert(name.clone(), value);
604 }
605 }
606 out
607}
608
609fn collect_generations(route: &Route, generations: &Generations) -> BTreeMap<String, String> {
611 let mut out = BTreeMap::new();
612 if let Some(reads) = &route.source_reads {
613 for name in reads {
614 out.insert(name.clone(), generations.token(name));
615 }
616 }
617 out
618}
619
620fn build_cached_response(entry: &CacheEntry, cache_label: &str, stale: bool) -> Response {
623 let mut builder = Response::builder()
624 .status(StatusCode::from_u16(entry.status).unwrap_or(StatusCode::OK));
625 let mut has_content_type = false;
626 for (k, v) in &entry.headers {
627 if k.eq_ignore_ascii_case("content-type") {
628 has_content_type = true;
629 }
630 builder = builder.header(k, v);
631 }
632 if !has_content_type {
633 builder = builder.header(header::CONTENT_TYPE, "text/html; charset=utf-8");
634 }
635 builder = builder.header("x-mesofact-cache", cache_label);
636 if stale {
637 builder = builder.header("x-mesofact-stale", "true");
638 }
639 builder.body(Body::from(entry.html.clone())).unwrap()
640}
641
642fn redirect_to_login(path: &str, query: &BTreeMap<String, String>, login_url: Option<&str>) -> Response {
643 let Some(login) = login_url else {
644 return Response::builder()
645 .status(StatusCode::UNAUTHORIZED)
646 .body(Body::from("401 Unauthorized — session required"))
647 .unwrap();
648 };
649 let original = if query.is_empty() {
650 path.to_string()
651 } else {
652 format!("{path}?{}", encode_query(query))
653 };
654 let sep = if login.contains('?') { '&' } else { '?' };
655 let target = format!("{login}{sep}next={}", percent_encode(&original));
656 Response::builder()
657 .status(StatusCode::FOUND)
658 .header(header::LOCATION, target)
659 .body(Body::empty())
660 .unwrap()
661}
662
663fn service_unavailable(retry_after: Duration) -> Response {
664 Response::builder()
665 .status(StatusCode::SERVICE_UNAVAILABLE)
666 .header(header::RETRY_AFTER, retry_after.as_secs().max(1).to_string())
667 .body(Body::from("503 Service Unavailable"))
668 .unwrap()
669}
670
671#[derive(Clone)]
682struct ErrorPages {
683 routes: Option<ErrorRoutes>,
684 fallback_dir: Option<PathBuf>,
685}
686
687impl ErrorPages {
688 fn from_state(st: &AppState) -> Self {
689 Self {
690 routes: st.manifest.error_routes.clone(),
691 fallback_dir: st.fallback_dir.clone(),
692 }
693 }
694
695 async fn not_found(&self) -> Response {
696 self.render(StatusCode::NOT_FOUND, "404 Not Found").await
697 }
698
699 async fn internal_error(&self) -> Response {
700 self.render(StatusCode::INTERNAL_SERVER_ERROR, "500 Internal Server Error")
701 .await
702 }
703
704 async fn render(&self, status: StatusCode, default_text: &'static str) -> Response {
708 if let Some(dir) = &self.fallback_dir {
709 if let Some(route) = self
710 .routes
711 .as_ref()
712 .and_then(|r| error_route_for(r, status))
713 {
714 for candidate in route_to_candidates(route) {
715 if let Ok(bytes) = tokio::fs::read(dir.join(&candidate)).await {
716 return Response::builder()
717 .status(status)
718 .header(header::CONTENT_TYPE, "text/html; charset=utf-8")
719 .body(Body::from(bytes))
720 .unwrap();
721 }
722 }
723 }
724 }
725 Response::builder()
726 .status(status)
727 .body(Body::from(default_text))
728 .unwrap()
729 }
730}
731
732fn error_route_for(routes: &ErrorRoutes, status: StatusCode) -> Option<&str> {
734 if status.as_u16() >= 500 {
735 routes.server_error.as_deref()
736 } else if status == StatusCode::NOT_FOUND {
737 routes.not_found.as_deref()
738 } else {
739 None
740 }
741}
742
743fn route_to_candidates(route: &str) -> Vec<PathBuf> {
746 let rel = route.trim_start_matches('/');
747 if rel.is_empty() {
748 return vec![PathBuf::from("index.html")];
749 }
750 let last = rel.rsplit('/').next().unwrap_or(rel);
751 if last.contains('.') {
752 vec![PathBuf::from(rel)]
753 } else {
754 vec![
755 PathBuf::from(format!("{rel}.html")),
756 PathBuf::from(rel).join("index.html"),
757 ]
758 }
759}
760
761fn parse_kv(query: Option<&str>) -> BTreeMap<String, String> {
766 let mut out = BTreeMap::new();
767 if let Some(q) = query {
768 for pair in q.split('&').filter(|p| !p.is_empty()) {
769 match pair.split_once('=') {
770 Some((k, v)) => out.insert(k.to_string(), v.to_string()),
771 None => out.insert(pair.to_string(), String::new()),
772 };
773 }
774 }
775 out
776}
777
778fn parse_cookies(headers: &HeaderMap) -> BTreeMap<String, String> {
779 let mut out = BTreeMap::new();
780 if let Some(raw) = headers.get(header::COOKIE).and_then(|v| v.to_str().ok()) {
781 for pair in raw.split(';') {
782 if let Some((k, v)) = pair.split_once('=') {
783 out.insert(k.trim().to_string(), v.trim().to_string());
784 }
785 }
786 }
787 out
788}
789
790fn encode_query(query: &BTreeMap<String, String>) -> String {
791 query
792 .iter()
793 .map(|(k, v)| format!("{k}={v}"))
794 .collect::<Vec<_>>()
795 .join("&")
796}
797
798fn percent_encode(s: &str) -> String {
802 let mut out = String::with_capacity(s.len());
803 for b in s.bytes() {
804 match b {
805 b'A'..=b'Z' | b'a'..=b'z' | b'0'..=b'9' | b'-' | b'_' | b'.' | b'~' | b'/' => {
806 out.push(b as char)
807 }
808 _ => out.push_str(&format!("%{b:02X}")),
809 }
810 }
811 out
812}
813
814#[cfg(test)]
815mod matchit_pattern_tests {
816 use super::to_matchit_pattern;
817
818 #[test]
819 fn translates_colon_params_and_star_wildcards() {
820 assert_eq!(to_matchit_pattern("/c/:slug"), "/c/{slug}");
821 assert_eq!(to_matchit_pattern("/api/users/:id"), "/api/users/{id}");
822 assert_eq!(to_matchit_pattern("/x/:a/y/:b"), "/x/{a}/y/{b}");
823 assert_eq!(to_matchit_pattern("/assets/*rest"), "/assets/{*rest}");
824 assert_eq!(to_matchit_pattern("/about"), "/about");
826 assert_eq!(to_matchit_pattern("/"), "/");
827 assert_eq!(to_matchit_pattern("/c/:"), "/c/:");
828 }
829
830 #[test]
834 fn translated_params_match_real_segments_and_bind_by_name() {
835 let mut r = matchit::Router::new();
836 r.insert(to_matchit_pattern("/c/:slug"), 1usize).unwrap();
837
838 let m = r.at("/c/hello").expect(":slug must match a real segment");
839 assert_eq!(*m.value, 1);
840 assert_eq!(m.params.get("slug"), Some("hello"));
841
842 assert!(r.at("/c/hello/extra").is_err());
844 }
845
846 #[test]
849 fn untranslated_colon_param_is_inert_under_matchit_08() {
850 let mut r = matchit::Router::new();
851 r.insert("/c/:slug", 1usize)
852 .expect("0.8 accepts ':' as a literal — this is why the bug was silent");
853 assert!(
854 r.at("/c/hello").is_err(),
855 "if this now matches, matchit restored ':' support and the shim can be revisited"
856 );
857 }
858}