1use std::collections::BTreeSet;
10use std::collections::HashMap;
11use std::sync::Arc;
12use std::time::Duration;
13
14use axum::Router;
15use axum::extract::{DefaultBodyLimit, State};
16use axum::http::{HeaderMap, StatusCode, header};
17use axum::middleware::{Next, from_fn_with_state};
18use axum::response::{IntoResponse, Response};
19use axum::routing::{get, post};
20use bytes::Bytes;
21use serde_json::json;
22use tracing::error;
23
24use crate::admin::require_admin;
25use crate::config::Config;
26use crate::config::ConfigError;
27use crate::events::EventBus;
28use crate::http::{
29 Json, MAX_BODY_BYTES, ScopeState, cors_layer, scope_layer, security_headers_layer,
30 token_response_layer,
31};
32use crate::module::{HARNESS_API, Module, ModuleContext, harness_api_mismatch};
33use crate::ports::Dispatcher;
34use crate::ports::{Clock, Database, Port, Ports, Statement, SystemClock, warn_undeclared_ports};
35use crate::problem::Problem;
36use crate::problems::SLUGS;
37use crate::route_policy::deployed_env;
38use crate::scope::Scope;
39use crate::sidecar::{
40 GatewayGuard, SIDECAR_REQUIRE_GATEWAY, SidecarMount, X_HARNESS_GATEWAY, gateway_guard,
41 gateway_signer, mint_gateway, truthy,
42};
43use crate::signer::HmacSigner;
44use crate::surface::{
45 MAX_SIDECAR_SURFACE_BYTES, RenderedSurface, SurfaceDocument, SurfaceSource, UiContext, UiMount,
46 sanitize_sidecar_document, strip_unguarded_captcha_actions,
47};
48use crate::template::{Template, TemplateRegistry};
49use crate::venture::{Venture, VentureEnv};
50
51fn check_tables(
60 module: &dyn Module,
61 tables: &mut HashMap<&'static str, &'static str>,
62 errors: &mut ConfigError,
63) {
64 let name = module.name();
65 for table in module.tables() {
66 match tables.get(table) {
67 Some(owner) => errors.push(format!(
68 "duplicate table `{table}` claimed by modules `{owner}` and `{name}`"
69 )),
70 None => {
71 tables.insert(table, name);
72 }
73 }
74 }
75
76 let owns = module.tables();
80 let mut declared: Vec<&'static str> = Vec::new();
81 for set in module.personal_data() {
82 for problem in set.validate(name, owns) {
83 errors.push(problem);
84 }
85 if declared.contains(&set.table) {
86 errors.push(format!(
87 "module `{name}` declares table `{}` twice in personal_data()",
88 set.table
89 ));
90 }
91 declared.push(set.table);
92 }
93}
94
95pub trait Runtime: Send + Sync + 'static {
100 fn provides(&self) -> Vec<Port>;
101 fn effectively_configured(&self, port: Port) -> bool {
107 self.provides().contains(&port)
108 }
109}
110
111pub struct Harness {
113 venture: Arc<Venture>,
114 modules: Vec<Arc<dyn Module>>,
115 templates: Arc<TemplateRegistry>,
116 events: EventBus,
117 runtime: Option<Arc<dyn Runtime>>,
118 well_known: Option<Router>,
121 surface: Arc<SurfaceVariants>,
124 ui: Option<Arc<dyn UiMount>>,
126 personal_data: Arc<crate::PersonalDataCatalog>,
129}
130
131struct SurfaceVariants {
132 document: Arc<SurfaceDocument>,
133 full: RenderedSurface,
134 public: RenderedSurface,
135}
136
137impl SurfaceVariants {
138 fn compose(
139 venture: &Venture,
140 modules: &[Arc<dyn Module>],
141 ui: Option<&Arc<dyn UiMount>>,
142 ) -> Self {
143 let mut document = SurfaceDocument::compose(venture, modules);
144 document.ui = ui.and_then(|ui| ui.describe());
145 Self {
146 full: RenderedSurface::render(&document),
147 public: RenderedSurface::render(&document.public()),
148 document: Arc::new(document),
149 }
150 }
151}
152
153impl std::fmt::Debug for Harness {
154 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
155 let modules: Vec<&str> = self.modules.iter().map(|m| m.name()).collect();
156 f.debug_struct("Harness")
157 .field("venture", &self.venture.name)
158 .field("modules", &modules)
159 .finish_non_exhaustive()
160 }
161}
162
163impl Harness {
164 #[must_use]
168 pub fn clone_arc(self: &std::sync::Arc<Self>) -> std::sync::Arc<Self> {
169 std::sync::Arc::clone(self)
170 }
171 pub fn builder() -> HarnessBuilder {
172 HarnessBuilder::default()
173 }
174
175 pub fn venture(&self) -> &Arc<Venture> {
176 &self.venture
177 }
178
179 pub fn modules(&self) -> &[Arc<dyn Module>] {
180 &self.modules
181 }
182
183 pub fn templates(&self) -> &Arc<TemplateRegistry> {
184 &self.templates
185 }
186
187 pub fn events(&self) -> &EventBus {
188 &self.events
189 }
190
191 pub fn module_context(&self, module: &dyn Module, ports: &Ports) -> ModuleContext {
196 let venture = self.venture_as_deployed(ports.config.as_ref());
211 ModuleContext {
212 config: Arc::clone(&ports.config),
213 ports: ports.view_for(module),
214 events: self.events_for(ports),
215 templates: Arc::clone(&self.templates),
216 venture,
217 unprotected_writes_accepted: crate::route_policy::unprotected_writes_override(
218 ports.config.as_ref(),
219 )
220 .is_some(),
221 personal_data: Arc::clone(&self.personal_data),
222 ui_mounted: self.ui.is_some(),
223 }
224 }
225
226 fn gateway_guard_state(ports: &Ports, gateway: Option<&Arc<HmacSigner>>) -> GatewayGuard {
236 GatewayGuard {
237 require: truthy(ports.config.as_ref(), SIDECAR_REQUIRE_GATEWAY),
238 signer: gateway.map(Arc::clone),
239 admin_token: ports
240 .config
241 .get("ADMIN_TOKEN")
242 .filter(|token| !token.is_empty()),
243 }
244 }
245
246 fn merged_surface(
249 &self,
250 mounts: Vec<SidecarMount>,
251 ports: &Ports,
252 gateway: Option<Arc<HmacSigner>>,
253 env: VentureEnv,
254 ) -> Arc<dyn SurfaceSource> {
255 Arc::new(MergedSurface {
256 base: Arc::clone(&self.surface),
257 mounts,
258 dispatcher: ports.dispatcher.clone(),
259 gateway,
260 env,
261 captcha_configured: ports.captcha.is_some(),
262 })
263 }
264
265 fn venture_as_deployed(&self, config: &dyn Config) -> Arc<Venture> {
271 let env = deployed_env(self.venture.env, config);
272 if env == self.venture.env {
273 return Arc::clone(&self.venture);
274 }
275 let mut deployed = (*self.venture).clone();
276 deployed.env = env;
277 Arc::new(deployed)
278 }
279
280 fn nest_modules(&self, ports: &Ports) -> Router {
283 let mut api = Router::new();
284 for module in &self.modules {
285 let ctx = self.module_context(module.as_ref(), ports);
286 api = api.nest(&format!("/v1/{}", module.name()), module.router(ctx));
287 }
288 api
289 }
290
291 fn production_readiness_now(&self, env: VentureEnv, config: &dyn Config) -> Vec<String> {
300 if let Some(note) = crate::route_policy::env_disagreement(self.venture.env, env) {
301 tracing::warn!("{note}");
302 }
303 let problems = crate::route_policy::production_readiness(
304 env,
305 &crate::route_policy::WriteGuards::collect(&self.modules),
306 self.runtime.as_ref(),
307 None,
308 );
309 if problems.is_empty() {
310 return problems;
311 }
312 if let Some(reason) = crate::route_policy::unprotected_writes_override(config) {
317 tracing::warn!(
318 control = "production-readiness",
319 reason,
320 problems = problems.join("; "),
321 "serving guarded routes unprotected on an operator's recorded acceptance"
322 );
323 return Vec::new();
324 }
325 for problem in &problems {
326 error!(
327 "refusing guarded routes: {problem} — set {} to a reason to accept this \
328 explicitly while the port is wired",
329 crate::route_policy::ALLOW_UNPROTECTED_WRITES
330 );
331 }
332 problems
333 }
334
335 fn nest_sidecars(
338 mut api: Router,
339 mounted: &[SidecarMount],
340 ports: &Ports,
341 gateway: Option<&Arc<HmacSigner>>,
342 ) -> Router {
343 for mount in mounted {
344 api = api.nest(
345 &format!("/v1/{}", mount.name),
346 crate::sidecar::router(
347 mount.clone(),
348 ports.dispatcher.clone(),
349 Arc::clone(&ports.config),
350 ports.rate_limiter.clone(),
351 gateway.map(Arc::clone),
352 ),
353 );
354 }
355 api
356 }
357
358 fn sidecar_mounts(&self, ports: &Ports) -> Vec<SidecarMount> {
359 let mounts = match crate::sidecar::SidecarMounts::from_config(ports.config.as_ref()) {
360 Ok(mounts) => mounts,
361 Err(errors) => {
362 for error in errors {
363 tracing::error!(error, "ignoring the sidecar mount table");
364 }
365 crate::sidecar::SidecarMounts::default()
366 }
367 };
368 let module_names: Vec<&str> = self.modules.iter().map(|m| m.name()).collect();
369 for collision in mounts.collisions(&module_names) {
370 tracing::error!(error = collision, "ignoring the colliding sidecar mount");
371 }
372 mounts
373 .iter()
374 .filter(|m| !module_names.contains(&m.name.as_str()))
375 .cloned()
376 .collect()
377 }
378
379 fn resolved_mounts(&self, ports: &Ports) -> Vec<SidecarMount> {
383 let mounts =
384 crate::sidecar::SidecarMounts::from_config(ports.config.as_ref()).unwrap_or_default();
385 let module_names: Vec<&str> = self.modules.iter().map(|m| m.name()).collect();
386 mounts
387 .iter()
388 .filter(|m| !module_names.contains(&m.name.as_str()))
389 .cloned()
390 .collect()
391 }
392
393 fn events_for(&self, ports: &Ports) -> EventBus {
401 let mounts = self.resolved_mounts(ports);
402 if mounts.is_empty() {
403 return self.events.clone();
404 }
405 let Some(dispatcher) = ports.dispatcher.clone() else {
406 return self.events.clone();
407 };
408 let gateway = gateway_signer(
409 ports.config.as_ref(),
410 ports.clock.clone().unwrap_or_else(|| Arc::new(SystemClock)),
411 );
412 self.events
413 .clone()
414 .forwarding_to(crate::sidecar::EventForwarder::new(
415 mounts, dispatcher, gateway,
416 ))
417 }
418
419 fn ui_router(
424 &self,
425 api: &Router,
426 ports: &Ports,
427 surface: &Arc<dyn SurfaceSource>,
428 ) -> Option<Router> {
429 self.ui.as_ref().map(|ui| {
430 ui.router(UiContext {
431 surface: Arc::clone(surface),
432 api: api.clone(),
433 config: Arc::clone(&ports.config),
434 venture: Arc::clone(&self.venture),
435 captcha_configured: ports.captcha.is_some(),
436 signer: ports.signer.clone(),
437 rate_limiter: ports.rate_limiter.clone(),
438 })
439 .layer(DefaultBodyLimit::max(MAX_BODY_BYTES))
440 })
441 }
442
443 fn inbound_events_route(&self, gateway: Option<Arc<HmacSigner>>) -> Option<Router> {
457 gateway.as_ref()?;
458 Some(
459 Router::new()
460 .route(
461 "/__events",
462 post(crate::sidecar::events_inbound)
463 .layer(DefaultBodyLimit::max(MAX_BODY_BYTES)),
464 )
465 .with_state(Arc::new(crate::sidecar::InboundEvents {
466 bus: self.events.clone(),
467 gateway,
468 })),
469 )
470 }
471
472 pub fn runtime(&self) -> Option<&Arc<dyn Runtime>> {
474 self.runtime.as_ref()
475 }
476
477 #[must_use]
481 pub fn surface(&self) -> &SurfaceDocument {
482 &self.surface.document
483 }
484
485 pub fn router(&self, ports: Ports) -> Router {
494 let env = deployed_env(self.venture.env, ports.config.as_ref());
497 let readiness = self.production_readiness_now(env, ports.config.as_ref());
498 let mut api = self.nest_modules(&ports);
499 let gateway = gateway_signer(
502 ports.config.as_ref(),
503 ports.clock.clone().unwrap_or_else(|| Arc::new(SystemClock)),
504 );
505 let mounted = self.sidecar_mounts(&ports);
506 api = Self::nest_sidecars(api, &mounted, &ports, gateway.as_ref());
507 let gateway_state = Self::gateway_guard_state(&ports, gateway.as_ref());
508 let mounts_for_health = mounted.clone();
509 let tenant_layer = Arc::new(TenantLayer {
511 routing: ports.tenants.clone(),
512 fallback: ports.db.clone(),
513 });
514 let api = api
519 .layer(axum::middleware::from_fn_with_state(
520 tenant_layer,
521 resolve_tenant_layer,
522 ))
523 .layer(axum::middleware::from_fn(security_headers_layer))
524 .layer(DefaultBodyLimit::max(MAX_BODY_BYTES));
525
526 let surface_source = self.merged_surface(mounted, &ports, gateway.clone(), env);
527
528 let ui = self.ui_router(&api, &ports, &surface_source);
529
530 let health_state = HealthState {
531 venture: self.venture_as_deployed(ports.config.as_ref()),
536 modules: self.modules.clone(),
537 harness_build: ports.config.get("HARNESS_BUILD").filter(|b| !b.is_empty()),
538 mailer_configured: ports.mailer.is_some(),
539 captcha_configured: ports.captcha.is_some(),
540 sidecars: mounts_for_health,
541 dispatcher: ports.dispatcher.clone(),
542 clock: ports.clock.clone().unwrap_or_else(|| Arc::new(SystemClock)),
543 probe_cache: new_probe_cache(),
544 };
545 let Ports {
546 config,
547 db,
548 mailer: _,
549 captcha: _,
550 rate_limiter: _,
551 signer: _,
552 kv: _,
553 blob: _,
554 push: _,
555 payments: _,
556 realtime: _,
557 http: _,
558 clock,
559 id_gen,
560 defer,
561 dispatcher: _,
562 tenants: _,
563 } = ports;
564
565 let scope_state = ScopeState {
566 defer: defer.unwrap_or_else(|| Arc::new(crate::ports::NoopDefer)),
567 id_gen: id_gen.unwrap_or_else(|| Arc::new(crate::ports::UlidIdGen)),
568 module: (self.modules.len() == 1).then(|| self.modules[0].name()),
571 };
572 let ready_state = ReadyState {
573 db,
574 clock: clock.unwrap_or_else(|| Arc::new(SystemClock)),
575 };
576
577 let surface_state = SurfaceState {
578 config,
579 source: surface_source,
580 };
581
582 let root = Router::new()
583 .route("/__health", get(health_handler))
584 .with_state(health_state)
585 .route("/__ready", get(ready_handler))
586 .with_state(ready_state)
587 .route("/__surface", get(surface_handler))
588 .with_state(surface_state)
589 .merge(
592 self.inbound_events_route(gateway.clone())
593 .unwrap_or_default(),
594 )
595 .merge(api);
596 let root = match &self.well_known {
597 Some(well_known) => root.nest(
598 "/.well-known",
599 well_known
600 .clone()
601 .layer(DefaultBodyLimit::max(MAX_BODY_BYTES)),
602 ),
603 None => root,
604 };
605 let root = match ui {
606 Some(ui) => root.nest("/ui", ui),
607 None => root,
608 };
609
610 root.layer(from_fn_with_state(
619 Arc::new(readiness),
620 production_readiness_guard,
621 ))
622 .layer(from_fn_with_state(gateway_state, gateway_guard))
623 .layer(from_fn_with_state(scope_state, scope_layer))
624 .layer(axum::middleware::from_fn(token_response_layer))
625 .layer(cors_layer(&self.venture.cors_origins))
626 }
627}
628
629type ProbeCache = Option<(i64, Vec<serde_json::Value>)>;
632
633#[allow(clippy::disallowed_types)]
636fn new_probe_cache() -> Arc<std::sync::Mutex<ProbeCache>> {
637 Arc::new(std::sync::Mutex::new(None))
638}
639
640#[derive(Clone)]
641struct HealthState {
642 venture: Arc<Venture>,
643 modules: Vec<Arc<dyn Module>>,
644 harness_build: Option<String>,
646 mailer_configured: bool,
647 captcha_configured: bool,
648 sidecars: Vec<SidecarMount>,
649 dispatcher: Option<Arc<dyn Dispatcher>>,
650 clock: Arc<dyn Clock>,
651 #[allow(clippy::disallowed_types)]
659 probe_cache: Arc<std::sync::Mutex<ProbeCache>>,
660}
661
662const SIDECAR_PROBE_TTL_SECS: i64 = 30;
665
666const SIDECAR_PROBE_TIMEOUT: Duration = Duration::from_secs(2);
669
670async fn health_handler(State(state): State<HealthState>) -> impl IntoResponse {
671 let modules: Vec<serde_json::Value> = state
672 .modules
673 .iter()
674 .map(|module| {
675 json!({
676 "name": module.name(),
677 "version": module.version(),
678 "requires": module.requires().iter().map(Port::name).collect::<Vec<_>>(),
679 "optional": module.optional().iter().map(Port::name).collect::<Vec<_>>(),
680 "tables": module.tables(),
681 "emits": module.emits(),
682 })
683 })
684 .collect();
685 let sidecars = probe_sidecars(&state).await;
686 Json(json!({
687 "venture": state.venture.name,
688 "env": state.venture.env.as_str(),
689 "harness_api": HARNESS_API,
690 "harness_build": state.harness_build,
691 "mailer": if state.mailer_configured { "configured" } else { "not_configured" },
694 "captcha": if state.captcha_configured { "configured" } else { "absent" },
695 "modules": modules,
696 "sidecars": sidecars,
697 }))
698}
699
700async fn probe_sidecars(state: &HealthState) -> Vec<serde_json::Value> {
706 if state.sidecars.is_empty() {
707 return Vec::new();
708 }
709 let now_ms =
710 i64::try_from(state.clock.now().unix_timestamp_nanos() / 1_000_000).unwrap_or(i64::MAX);
711 if let Some((at_ms, entries)) = state.probe_cache.lock().unwrap().as_ref()
712 && now_ms - at_ms < SIDECAR_PROBE_TTL_SECS * 1000
713 {
714 return entries.clone();
715 }
716
717 let mut entries = Vec::new();
718 for mount in &state.sidecars {
719 entries.push(probe_one_sidecar(state, mount).await);
720 }
721 *state.probe_cache.lock().unwrap() = Some((now_ms, entries.clone()));
722 entries
723}
724
725async fn probe_one_sidecar(state: &HealthState, mount: &SidecarMount) -> serde_json::Value {
726 let mut missing = json!({
727 "name": mount.name,
728 "binding": mount.binding,
729 "probe": "unreachable",
730 "contract": serde_json::Value::Null,
731 "module": serde_json::Value::Null,
732 "version": serde_json::Value::Null,
733 });
734 let Some(dispatcher) = state.dispatcher.as_ref() else {
735 return missing;
736 };
737 let request = http::Request::builder()
738 .method(http::Method::GET)
739 .uri("/__health")
740 .body(Bytes::new());
741 let Ok(request) = request else {
742 return missing;
743 };
744 let dispatcher = Arc::clone(dispatcher);
749 let binding = mount.binding.clone();
750 let probe = async move { dispatcher.dispatch(&binding, request).await };
751 let Some(Ok(response)) =
752 crate::ports::timeout(&*state.clock, probe, SIDECAR_PROBE_TIMEOUT).await
753 else {
754 return missing;
755 };
756 let contract = response
757 .headers()
758 .get(crate::sidecar::X_HARNESS_API)
759 .and_then(|v| v.to_str().ok())
760 .and_then(|v| v.parse::<u32>().ok());
761 let body = serde_json::from_slice::<serde_json::Value>(response.body()).ok();
762 let module = response
763 .headers()
764 .get(crate::sidecar::X_HARNESS_MODULE)
765 .and_then(|v| v.to_str().ok())
766 .map(str::to_owned)
767 .or_else(|| first_module_string(body.as_ref(), "name"));
768 let version = first_module_string(body.as_ref(), "version");
769 let tables = declared_tables(body.as_ref());
770 let contract = contract.or_else(|| {
771 body.as_ref()
772 .and_then(|b| b.get("harness_api"))
773 .and_then(serde_json::Value::as_u64)
774 .and_then(|api| u32::try_from(api).ok())
775 });
776 missing["contract"] = contract.map_or(serde_json::Value::Null, |api| json!(api));
777 missing["module"] = module.map_or(serde_json::Value::Null, |name| json!(name));
778 missing["version"] = version.map_or(serde_json::Value::Null, |version| json!(version));
779 missing["tables"] = json!(tables);
780 match contract {
781 Some(api) if api != HARNESS_API => missing["probe"] = json!("mismatch"),
782 Some(_) => missing["probe"] = json!("ok"),
783 None => missing["probe"] = json!("unreachable"),
784 }
785 let clashes = collisions_with(&tables, state, &mount.name);
797 if !clashes.is_empty() {
798 let detail = format!(
799 "sidecar `{}` claims table(s) {} already claimed in this deployment",
800 mount.name,
801 clashes.join(", ")
802 );
803 tracing::error!(sidecar = %mount.name, tables = %clashes.join(","), "{detail}");
804 crate::logging::forward_internal_error(&detail);
805 missing["probe"] = json!("table-collision");
806 missing["table_collisions"] = json!(clashes);
807 }
808 missing
809}
810
811fn declared_tables(body: Option<&serde_json::Value>) -> Vec<String> {
815 let Some(modules) = body
816 .and_then(|b| b.get("modules"))
817 .and_then(|m| m.as_array())
818 else {
819 return Vec::new();
820 };
821 let mut tables: Vec<String> = modules
822 .iter()
823 .filter_map(|module| module.get("tables")?.as_array())
824 .flatten()
825 .filter_map(|table| table.as_str().map(str::to_owned))
826 .collect();
827 tables.sort_unstable();
828 tables.dedup();
829 tables
830}
831
832fn collisions_with(claimed: &[String], state: &HealthState, mount: &str) -> Vec<String> {
835 let mut clashes: Vec<String> = Vec::new();
836 for module in &state.modules {
837 if module.name() == mount {
838 continue;
839 }
840 for table in module.tables() {
841 if claimed.iter().any(|claimed| claimed == table)
842 && !clashes.iter().any(|seen| seen == table)
843 {
844 clashes.push(format!("`{table}` (also module `{}`)", module.name()));
845 }
846 }
847 }
848 clashes
849}
850
851fn first_module_string(body: Option<&serde_json::Value>, key: &str) -> Option<String> {
854 body?
855 .get("modules")?
856 .get(0)?
857 .get(key)?
858 .as_str()
859 .map(str::to_owned)
860}
861
862struct TenantLayer {
865 routing: Option<Arc<dyn crate::tenant::TenantRouting>>,
866 fallback: Option<Arc<dyn Database>>,
872}
873
874async fn resolve_tenant_layer(
881 State(layer): State<Arc<TenantLayer>>,
882 scope: Scope,
883 mut request: axum::extract::Request,
884 next: Next,
885) -> Response {
886 let host = request
887 .headers()
888 .get(header::HOST)
889 .and_then(|value| value.to_str().ok())
890 .unwrap_or_default()
891 .to_owned();
892
893 let resolution = match layer.routing.as_ref() {
898 Some(routing) => routing.resolve(&host),
899 None => <crate::tenant::ImplicitTenant as crate::tenant::ResolveTenant>::resolve(
900 &crate::tenant::ImplicitTenant,
901 &host,
902 ),
903 };
904
905 let tenant = match resolution.admit() {
906 Ok(tenant) => tenant,
907 Err(problem) => {
908 return Problem::new(problem)
909 .instance(&scope.request_id)
910 .into_response();
911 }
912 };
913
914 let db = match layer.routing.as_ref() {
915 Some(routing) => match routing.database(&tenant).await {
916 Ok(db) => Some(db),
917 Err(err) => {
918 tracing::error!(tenant = %tenant.id(), "{err}");
921 crate::logging::forward_internal_error(&err.to_string());
922 return Problem::new(&SLUGS.tenant_degraded)
923 .instance(&scope.request_id)
924 .into_response();
925 }
926 },
927 None => layer.fallback.clone(),
928 };
929
930 if let Some(db) = db {
931 request
932 .extensions_mut()
933 .insert(crate::tenant_conn::ResolvedTenant { tenant, db });
934 }
935 next.run(request).await
936}
937
938#[derive(Clone)]
939struct ReadyState {
940 db: Option<Arc<dyn Database>>,
941 clock: Arc<dyn Clock>,
942}
943
944async fn ready_handler(State(state): State<ReadyState>) -> impl IntoResponse {
948 let Some(db) = state.db else {
949 return Problem::not_ready("database port is not configured").into_response();
950 };
951 let stmt = Statement::new("SELECT 1");
952 let query = async move { db.query(&stmt).await };
953 match crate::ports::timeout(&*state.clock, query, Duration::from_secs(2)).await {
954 Some(Ok(_rows)) => Json(json!({ "ok": true })).into_response(),
955 Some(Err(err)) => {
956 error!(error = %err, "readiness probe query failed");
957 crate::logging::forward_internal_error(&format!("readiness probe query failed: {err}"));
958 Problem::not_ready("database query failed").into_response()
959 }
960 None => Problem::not_ready("database did not answer within 2 s").into_response(),
961 }
962}
963
964#[derive(Clone)]
965struct SurfaceState {
966 config: Arc<dyn Config>,
967 source: Arc<dyn SurfaceSource>,
968}
969
970struct MergedSurface {
977 base: Arc<SurfaceVariants>,
978 mounts: Vec<SidecarMount>,
979 dispatcher: Option<Arc<dyn Dispatcher>>,
980 gateway: Option<Arc<HmacSigner>>,
983 env: VentureEnv,
984 captcha_configured: bool,
985}
986
987impl MergedSurface {
988 async fn sidecar_modules(&self) -> Vec<crate::surface::ModuleSurface> {
990 let mut extra = Vec::new();
991 let Some(dispatcher) = &self.dispatcher else {
992 return extra;
993 };
994 for mount in &self.mounts {
995 if !dispatcher.has(&mount.binding) {
996 continue;
997 }
998 let mut builder = axum::http::Request::builder()
999 .method(axum::http::Method::GET)
1000 .uri("/__surface")
1001 .header(header::ACCEPT, "application/json");
1002 if let Some(signer) = &self.gateway {
1003 builder =
1005 builder.header(X_HARNESS_GATEWAY, mint_gateway(signer, &mount.name, false));
1006 }
1007 let request = builder
1008 .body(bytes::Bytes::new())
1009 .expect("static request builds");
1010 let answer = match dispatcher.dispatch(&mount.binding, request).await {
1011 Ok(response) if response.status().is_success() => response,
1012 Ok(response) => {
1013 tracing::warn!(module = mount.name, status = %response.status(), "sidecar surface not available");
1014 continue;
1015 }
1016 Err(err) => {
1017 tracing::warn!(module = mount.name, error = %err, "sidecar surface fetch failed");
1018 continue;
1019 }
1020 };
1021 if answer.body().len() > MAX_SIDECAR_SURFACE_BYTES {
1024 tracing::warn!(
1025 module = mount.name,
1026 "sidecar surface exceeds {MAX_SIDECAR_SURFACE_BYTES} bytes and was not merged"
1027 );
1028 continue;
1029 }
1030 match serde_json::from_slice::<SurfaceDocument>(answer.body()) {
1031 Ok(document) => match sanitize_sidecar_document(&document, &mount.name) {
1032 Ok(entries) => extra.extend(entries.into_iter().filter_map(|mut entry| {
1033 if self.env == VentureEnv::Production
1037 && !self.captcha_configured
1038 && strip_unguarded_captcha_actions(&mut entry.surface) > 0
1039 {
1040 tracing::warn!(
1041 module = entry.name,
1042 "merged sidecar actions requiring a captcha were dropped: no Captcha port here"
1043 );
1044 }
1045 (!entry.surface.is_empty()).then_some(entry)
1046 })),
1047 Err(problems) => {
1048 for problem in problems {
1049 tracing::warn!(module = mount.name, error = problem, "sidecar surface rejected at merge");
1050 }
1051 }
1052 },
1053 Err(err) => {
1054 tracing::warn!(module = mount.name, error = %err, "sidecar surface is not a surface document");
1055 }
1056 }
1057 }
1058 extra
1059 }
1060}
1061
1062#[async_trait::async_trait]
1063impl SurfaceSource for MergedSurface {
1064 async fn current(&self) -> Arc<SurfaceDocument> {
1065 if self.mounts.is_empty() {
1066 return Arc::clone(&self.base.document);
1067 }
1068 let mut document = (*self.base.document).clone();
1069 document.modules.extend(self.sidecar_modules().await);
1070 Arc::new(document)
1071 }
1072
1073 fn built(&self) -> Arc<SurfaceDocument> {
1074 Arc::clone(&self.base.document)
1075 }
1076
1077 fn rendered(&self, admin: bool) -> Option<&RenderedSurface> {
1078 Some(if admin {
1079 &self.base.full
1080 } else {
1081 &self.base.public
1082 })
1083 }
1084}
1085
1086async fn surface_handler(
1102 State(state): State<SurfaceState>,
1103 headers: HeaderMap,
1104) -> impl IntoResponse {
1105 let admin = require_admin(&*state.config, &headers).is_ok();
1106 let current = state.source.current().await;
1110 let built = state.source.built();
1111 let prerendered = Arc::ptr_eq(¤t, &built).then(|| state.source.rendered(admin));
1112 let fresh;
1113 let rendered: &RenderedSurface = if let Some(rendered) = prerendered.flatten() {
1114 rendered
1115 } else {
1116 fresh = if admin {
1117 RenderedSurface::render(¤t)
1118 } else {
1119 RenderedSurface::render(¤t.public())
1120 };
1121 &fresh
1122 };
1123 let matches = headers
1124 .get(header::IF_NONE_MATCH)
1125 .and_then(|value| value.to_str().ok())
1126 .is_some_and(|value| {
1127 value
1128 .split(',')
1129 .map(str::trim)
1130 .any(|tag| tag == "*" || tag == rendered.etag)
1131 });
1132 let mut response = if matches {
1133 StatusCode::NOT_MODIFIED.into_response()
1134 } else {
1135 (
1136 [(header::CONTENT_TYPE, "application/json")],
1137 rendered.json.clone(),
1138 )
1139 .into_response()
1140 };
1141 let response_headers = response.headers_mut();
1142 response_headers.insert(
1143 header::ETAG,
1144 header::HeaderValue::from_str(&rendered.etag).expect("hex etag is a valid header"),
1145 );
1146 response_headers.insert(
1147 header::CACHE_CONTROL,
1148 if admin {
1149 header::HeaderValue::from_static("private, no-store")
1150 } else {
1151 header::HeaderValue::from_static("no-cache")
1152 },
1153 );
1154 response_headers.insert(
1155 header::VARY,
1156 header::HeaderValue::from_static("Authorization"),
1157 );
1158 response
1159}
1160
1161#[derive(Default)]
1164pub struct HarnessBuilder {
1165 venture: Option<Venture>,
1166 modules: Vec<Arc<dyn Module>>,
1167 provides: Vec<Port>,
1168 runtime: Option<Arc<dyn Runtime>>,
1169 module_templates: Vec<(String, Box<dyn Template>)>,
1170 overrides: Vec<(String, Box<dyn Template>)>,
1171 ui: Option<Arc<dyn UiMount>>,
1172}
1173
1174impl HarnessBuilder {
1175 #[must_use]
1176 pub fn venture(mut self, venture: Venture) -> Self {
1177 self.venture = Some(venture);
1178 self
1179 }
1180
1181 #[must_use]
1184 pub fn module(mut self, module: impl Module) -> Self {
1185 self.modules.push(Arc::new(module));
1186 self
1187 }
1188
1189 #[must_use]
1192 pub fn module_arc(mut self, module: Arc<dyn Module>) -> Self {
1193 self.modules.push(module);
1194 self
1195 }
1196
1197 #[must_use]
1201 pub fn runtime(mut self, runtime: impl Runtime) -> Self {
1202 let runtime: Arc<dyn Runtime> = Arc::new(runtime);
1203 self.provides = runtime.provides();
1204 self.runtime = Some(runtime);
1205 self
1206 }
1207
1208 #[must_use]
1211 pub fn templates(
1212 mut self,
1213 templates: impl IntoIterator<Item = (String, Box<dyn Template>)>,
1214 ) -> Self {
1215 self.module_templates.extend(templates);
1216 self
1217 }
1218
1219 #[must_use]
1222 pub fn ui(mut self, ui: impl UiMount) -> Self {
1223 self.ui = Some(Arc::new(ui));
1224 self
1225 }
1226
1227 #[must_use]
1230 pub fn template(mut self, id: impl Into<String>, template: Box<dyn Template>) -> Self {
1231 self.overrides.push((id.into(), template));
1232 self
1233 }
1234
1235 pub fn build(self) -> Result<Harness, ConfigError> {
1246 let mut errors = ConfigError::default();
1247
1248 let venture = if let Some(venture) = self.venture {
1249 venture.validate(&mut errors);
1250 venture
1251 } else {
1252 errors.push("missing venture: call .venture(Venture::new(..)) before .build()");
1253 Venture::new("invalid", "invalid.invalid")
1254 };
1255
1256 let mut names: HashMap<&'static str, usize> = HashMap::new();
1257 let mut tables: HashMap<&'static str, &'static str> = HashMap::new();
1258
1259 for module in &self.modules {
1260 if module.harness_api() != HARNESS_API {
1261 errors.push(harness_api_mismatch(module.as_ref()));
1262 }
1263
1264 let name = module.name();
1265 if name.is_empty() || !is_module_name(name) {
1266 errors.push(format!(
1267 "module name `{name}` must be kebab-case ([a-z0-9]+ separated by '-')"
1268 ));
1269 }
1270 if names.insert(name, 1).is_some() {
1271 errors.push(format!("duplicate module name `{name}`"));
1272 }
1273
1274 for port in module.requires().iter().chain(module.optional()) {
1275 if !Port::ALL.contains(port) {
1276 errors.push(format!(
1277 "module `{name}` declares unknown port {}",
1278 port.name()
1279 ));
1280 }
1281 }
1282 for port in module.requires() {
1283 if module.optional().contains(port) {
1284 errors.push(format!(
1285 "module `{name}` lists port {} in both requires() and optional()",
1286 port.name()
1287 ));
1288 }
1289 }
1290
1291 module.surface().validate(name, &mut errors);
1292
1293 check_tables(module.as_ref(), &mut tables, &mut errors);
1294 }
1295
1296 let well_known = collect_well_known(&self.modules, &mut errors);
1297
1298 for module in &self.modules {
1299 for port in module.requires() {
1300 if !self.provides.contains(port) {
1301 errors.push(format!(
1302 "module `{}` requires port {} which the runtime does not provide",
1303 module.name(),
1304 port.name()
1305 ));
1306 }
1307 }
1308 warn_undeclared_ports(module.as_ref(), &self.provides);
1309 }
1310
1311 let ordered = resolve_dependency_order(&self.modules, &mut errors);
1312
1313 append_production_readiness(&venture, &self.modules, self.runtime.as_ref(), &mut errors);
1314
1315 check_template_ids(
1316 self.overrides.iter().chain(self.module_templates.iter()),
1317 &names,
1318 &mut errors,
1319 );
1320
1321 let surface = Arc::new(SurfaceVariants::compose(
1322 &venture,
1323 &self.modules,
1324 self.ui.as_ref(),
1325 ));
1326 if let Some(ui) = &self.ui {
1327 ui.validate(&surface.document, &mut errors);
1328 }
1329
1330 errors.into_result()?;
1331
1332 let mut registry = TemplateRegistry::new();
1333 registry.register_all(self.module_templates);
1334 registry.register_all(self.overrides);
1335
1336 let mut events = EventBus::new();
1337 for module in &self.modules {
1338 for (name, handler) in module.events() {
1339 events = events.on(name, handler);
1340 }
1341 }
1342
1343 Ok(Harness {
1344 venture: Arc::new(venture),
1345 personal_data: Arc::new(crate::PersonalDataCatalog::compose(
1351 ordered.iter().map(|m| (m.name(), m.personal_data())),
1352 )),
1353 modules: ordered,
1354 templates: Arc::new(registry),
1355 events,
1356 runtime: self.runtime,
1357 well_known,
1358 surface,
1359 ui: self.ui,
1360 })
1361 }
1362}
1363
1364fn resolve_dependency_order(
1377 modules: &[Arc<dyn Module>],
1378 errors: &mut ConfigError,
1379) -> Vec<Arc<dyn Module>> {
1380 let names: BTreeSet<&str> = modules.iter().map(|module| module.name()).collect();
1381 let mut unknown = false;
1382 for module in modules {
1383 for needed in module.depends_on() {
1384 if !names.contains(needed) {
1385 unknown = true;
1386 errors.push(format!(
1387 "module `{}` depends on `{needed}`, which this venture does not compose: \
1388 add the module or drop the dependency",
1389 module.name()
1390 ));
1391 }
1392 }
1393 }
1394 if unknown {
1395 return modules.to_vec();
1396 }
1397
1398 let mut remaining: Vec<Option<Arc<dyn Module>>> = modules
1402 .iter()
1403 .map(|module| Some(Arc::clone(module)))
1404 .collect();
1405 let mut placed: BTreeSet<&str> = BTreeSet::new();
1406 let mut ordered: Vec<Arc<dyn Module>> = Vec::with_capacity(modules.len());
1407
1408 while ordered.len() < modules.len() {
1409 let mut progressed = false;
1410 for slot in &mut remaining {
1411 let ready = slot.as_ref().is_some_and(|module| {
1412 module
1413 .depends_on()
1414 .iter()
1415 .all(|needed| placed.contains(needed))
1416 });
1417 if ready {
1418 let module = slot.take().expect("checked above");
1419 placed.insert(module.name());
1420 ordered.push(module);
1421 progressed = true;
1422 }
1423 }
1424 if !progressed {
1425 let stuck: Vec<&str> = remaining
1427 .iter()
1428 .filter_map(|slot| slot.as_ref().map(|module| module.name()))
1429 .collect();
1430 errors.push(format!(
1431 "modules [{}] depend on each other in a cycle: migrations cannot be ordered so \
1432 that every module follows the ones it sits on top of",
1433 stuck.join(", ")
1434 ));
1435 return modules.to_vec();
1436 }
1437 }
1438 ordered
1439}
1440
1441fn check_template_ids<'a>(
1442 ids: impl Iterator<Item = &'a (String, Box<dyn Template>)>,
1443 names: &HashMap<&'static str, usize>,
1444 errors: &mut ConfigError,
1445) {
1446 for (id, _) in ids {
1447 let Some(module_name) = id.split('/').next() else {
1448 continue;
1449 };
1450 if !names.contains_key(module_name) {
1451 errors.push(format!(
1452 "template `{id}` names module `{module_name}` which is not registered"
1453 ));
1454 }
1455 }
1456}
1457
1458fn is_module_name(name: &str) -> bool {
1459 name.split('-').all(|part| {
1460 !part.is_empty()
1461 && part
1462 .chars()
1463 .all(|c| c.is_ascii_lowercase() || c.is_ascii_digit())
1464 })
1465}
1466
1467async fn production_readiness_guard(
1477 State(problems): State<Arc<Vec<String>>>,
1478 request: axum::extract::Request,
1479 next: Next,
1480) -> Response {
1481 if problems.is_empty() || !request.uri().path().starts_with("/v1/") {
1482 return next.run(request).await;
1483 }
1484 let problem = Problem::new(&SLUGS.not_production_ready).with_detail(problems.join("; "));
1485 match request.extensions().get::<Scope>() {
1486 Some(scope) => problem.instance(&scope.request_id.clone()),
1487 None => problem,
1488 }
1489 .into_response()
1490}
1491
1492fn append_production_readiness(
1502 venture: &Venture,
1503 modules: &[Arc<dyn Module>],
1504 runtime: Option<&Arc<dyn Runtime>>,
1505 errors: &mut ConfigError,
1506) {
1507 for error in crate::route_policy::production_readiness(
1508 venture.env,
1509 &crate::route_policy::WriteGuards::collect(modules),
1510 runtime,
1511 None,
1512 ) {
1513 errors.push(error);
1514 }
1515}
1516
1517fn collect_well_known(modules: &[Arc<dyn Module>], errors: &mut ConfigError) -> Option<Router> {
1518 let mut providers: Vec<&'static str> = Vec::new();
1519 let mut well_known = None;
1520 for module in modules {
1521 if let Some(router) = module.well_known() {
1522 providers.push(module.name());
1523 well_known = Some(router);
1524 }
1525 }
1526 if providers.len() > 1 {
1527 let listed = providers
1528 .iter()
1529 .map(|name| format!("`{name}`"))
1530 .collect::<Vec<_>>()
1531 .join(", ");
1532 errors.push(format!(
1533 "modules {listed} all provide a well-known router; at most one module may occupy /.well-known"
1534 ));
1535 }
1536 well_known
1537}