1#[cfg(feature = "handlers")]
10use crate::error::Error;
11use crate::error::Result;
12use boatramp_core::deploy::DeployStore;
13use boatramp_core::envelope::KeyEnvelope;
14use boatramp_core::kv::KvStore;
15use std::path::Path;
16use std::sync::Arc;
17
18#[cfg(feature = "handlers")]
24const DEFAULT_ASYNC_TIMEOUT_MS: u64 = 15 * 60 * 1000;
25
26#[cfg(feature = "handlers")]
30const DEFAULT_ASYNC_CONCURRENCY: usize = 8;
31
32#[cfg(feature = "handlers")]
35const DEFAULT_STREAMING_TIMEOUT_MS: u64 = 15 * 60 * 1000;
36
37#[cfg(feature = "handlers")]
41const DEFAULT_STREAMING_CONCURRENCY: usize = 64;
42
43#[cfg(feature = "handlers")]
47#[allow(clippy::too_many_arguments)]
48pub async fn build_handler_runtime(
49 kv: Arc<dyn KvStore>,
50 storage: Arc<dyn boatramp_core::Storage>,
51 data_dir: &Path,
52 handlers_cfg: Option<&crate::config::HandlersConfig>,
53 messaging_override: Option<Arc<dyn boatramp_core::messaging::Messaging>>,
54 max_blob_bytes: u64,
55 max_component_bytes: u64,
56 allow_guest_private_egress: bool,
58 self_egress_addrs: Vec<std::net::SocketAddr>,
60 guest_egress_extra_roots: Vec<rustls::pki_types::CertificateDer<'static>>,
64 allow_env_secret_refs: bool,
68 allow_guest_email: bool,
72 require_tenancy_declaration: bool,
76 allow_cross_tenant_db: bool,
77 deploy: &DeployStore,
80 secrets_envelope: Option<Arc<dyn KeyEnvelope>>,
81) -> Result<boatramp_server::HandlerRuntime> {
82 #[cfg(not(feature = "email"))]
85 let _ = allow_guest_email;
86 let defaults = boatramp_handlers::Limits::default();
96 let sync_limits = boatramp_handlers::Limits {
97 timeout_ms: handlers_cfg
98 .and_then(|h| h.sync_max_timeout_ms)
99 .unwrap_or(defaults.timeout_ms),
100 ..defaults
101 };
102 let async_limits = boatramp_handlers::Limits {
103 timeout_ms: handlers_cfg
104 .and_then(|h| h.async_max_timeout_ms)
105 .unwrap_or(DEFAULT_ASYNC_TIMEOUT_MS),
106 max_concurrency: handlers_cfg
107 .and_then(|h| h.async_max_concurrency)
108 .unwrap_or(DEFAULT_ASYNC_CONCURRENCY),
109 fuel: handlers_cfg.and_then(|h| h.async_max_fuel),
110 ..sync_limits
111 };
112 let streaming_limits = boatramp_handlers::Limits {
117 timeout_ms: handlers_cfg
118 .and_then(|h| h.streaming_max_timeout_ms)
119 .unwrap_or(DEFAULT_STREAMING_TIMEOUT_MS),
120 max_concurrency: handlers_cfg
121 .and_then(|h| h.streaming_max_concurrency)
122 .unwrap_or(DEFAULT_STREAMING_CONCURRENCY),
123 fuel: handlers_cfg.and_then(|h| h.streaming_max_fuel),
124 ..sync_limits
125 };
126 let outbound_timeout = handlers_cfg
127 .and_then(|h| h.outbound_timeout_ms)
128 .map(std::time::Duration::from_millis);
129 let engine = if handlers_cfg.is_some_and(|h| h.pooling) {
132 boatramp_handlers::HandlerEngine::with_pooling(sync_limits, 64)?
133 } else {
134 boatramp_handlers::HandlerEngine::new(sync_limits, 64)?
135 }
136 .with_async_limits(async_limits)
137 .with_streaming_limits(streaming_limits)
138 .with_outbound_timeout(outbound_timeout)
139 .with_private_egress(allow_guest_private_egress)
140 .with_self_egress(self_egress_addrs)
141 .with_guest_egress_extra_roots(guest_egress_extra_roots);
142 let sql = build_sql_backends(
143 handlers_cfg.and_then(|h| h.bindings.sql.as_ref()),
144 data_dir,
145 deploy,
146 &kv,
147 secrets_envelope.as_ref(),
148 )
149 .await?;
150 let max_unflushed = handlers_cfg
157 .and_then(|h| h.messaging_max_unflushed_msgs)
158 .unwrap_or(0);
159 if max_unflushed > 0 {
160 tracing::warn!(
161 max_unflushed,
162 "messaging: RELAXED publish durability ENABLED (messaging_max_unflushed_msgs={max_unflushed}) \
163 — publish() acks before the WAL flush; up to {max_unflushed} acknowledged-but-unflushed \
164 messages are lost on a process crash / OOM / SIGKILL / power loss. Control-plane \
165 durability is UNAFFECTED. Set 0 (the default) for stronger-than-JetStream durability."
166 );
167 }
168 let messaging: Arc<dyn boatramp_core::messaging::Messaging> = messaging_override
169 .unwrap_or_else(|| {
170 Arc::new(
171 boatramp_core::messaging::LogMessaging::new(storage.clone(), kv.clone())
172 .with_max_unflushed(max_unflushed),
173 )
174 });
175 let kv_for_secrets = kv.clone();
178 #[cfg(feature = "email")]
182 let kv_for_email = kv.clone();
183 #[cfg(feature = "email")]
184 let messaging_for_email = messaging.clone();
185 #[cfg(feature = "email")]
186 let email_envelope = secrets_envelope.clone();
187 let runtime =
188 boatramp_server::HandlerRuntime::new(engine, kv, storage, Some(sql), Some(messaging));
189 runtime.set_max_blob_bytes(max_blob_bytes);
191 runtime.set_max_component_bytes(max_component_bytes);
192 runtime.set_allow_env_secret_refs(allow_env_secret_refs);
194 runtime.set_tenancy_posture(require_tenancy_declaration, allow_cross_tenant_db);
196 if let Some(envelope) = secrets_envelope {
200 runtime.set_secret_store(Arc::new(boatramp_core::secret_store::SecretStore::new(
201 kv_for_secrets,
202 envelope,
203 )));
204 }
205 #[cfg(feature = "email")]
213 if allow_guest_email {
214 if let Some(envelope) = email_envelope {
215 let store = Arc::new(boatramp_core::email_config::EmailProfileStore::new(
216 kv_for_email,
217 envelope,
218 ));
219 let backend = Arc::new(boatramp_handlers::LettreBackend::new(
220 allow_guest_private_egress,
221 ));
222 let spool = boatramp_server::NodeEmailSpool::spawn(
223 backend,
224 Some(messaging_for_email),
225 store.clone(),
226 );
227 runtime.set_email_profile_store(store);
228 runtime.set_email_spool(spool);
229 }
230 }
231 Ok(runtime)
232}
233
234#[cfg(feature = "handlers")]
239async fn build_sql_backends(
240 cfg: Option<&crate::config::SqlBindingConfig>,
241 data_dir: &Path,
242 deploy: &DeployStore,
243 kv: &Arc<dyn KvStore>,
244 secrets_envelope: Option<&Arc<dyn KeyEnvelope>>,
245) -> Result<Arc<dyn boatramp_core::sql::SqlBackends>> {
246 let resolve_env = |var: &Option<String>| -> Result<Option<String>> {
247 match var {
248 Some(var) => Ok(Some(
249 std::env::var(var).map_err(|_| Error::SqlEnvUnset(var.clone()))?,
250 )),
251 None => Ok(None),
252 }
253 };
254
255 let backend = match cfg.and_then(|c| c.url.as_ref()) {
256 Some(url) => {
259 let cfg = cfg.expect("url implies cfg");
260 let admin_url = cfg.admin_url.as_ref().ok_or(Error::SqlAdminUrlRequired)?;
261 let token = resolve_env(&cfg.token_env)?.unwrap_or_default();
262 let admin_token = resolve_env(&cfg.admin_token_env)?;
263 let backends = boatramp_storage::LibsqlSqlBackends::remote(
264 url.clone(),
265 admin_url.clone(),
266 token,
267 admin_token,
268 );
269 match &cfg.replica_url {
271 Some(replica_url) => backends.with_read_replica(replica_url.clone()),
272 None => backends,
273 }
274 }
275 None => {
277 let dir = cfg
278 .and_then(|c| c.dir.clone())
279 .unwrap_or_else(|| data_dir.join("handlers-sql"));
280 boatramp_storage::LibsqlSqlBackends::local(dir)
281 }
282 };
283 let preview_mode = match cfg.and_then(|c| c.preview_mode.as_deref()) {
285 None | Some("empty") => boatramp_core::sql::PreviewSqlMode::Empty,
286 Some("branch") => boatramp_core::sql::PreviewSqlMode::Branch,
287 Some("shared") => boatramp_core::sql::PreviewSqlMode::Shared,
288 Some(other) => return Err(Error::UnknownPreviewMode(other.to_string())),
289 };
290 let preview_init = match cfg.and_then(|c| c.preview_init.as_ref()) {
291 Some(path) => {
292 Some(
293 std::fs::read_to_string(path).map_err(|err| Error::PreviewInitRead {
294 path: path.clone(),
295 source: err,
296 })?,
297 )
298 }
299 None => None,
300 };
301 let default: Arc<dyn boatramp_core::sql::SqlBackends> =
302 Arc::new(backend.with_preview_policy(preview_mode, preview_init));
303
304 let databases = cfg.map(|c| &c.databases);
308 if databases.is_none_or(std::collections::BTreeMap::is_empty) {
309 return Ok(default);
310 }
311 let databases = databases.expect("checked non-empty above");
312
313 #[cfg(any(feature = "sql-postgres", feature = "sql-mysql"))]
314 {
315 use boatramp_core::sql::SqlBackend;
316 use boatramp_storage::sql_sqlx::{
317 connect, CompositeSqlBackends, ExternalSqlKind, ExternalSqlOptions,
318 };
319 let timeout = |db: &crate::config::ExternalDatabaseConfig| {
320 db.connect_timeout_secs.map(std::time::Duration::from_secs)
321 };
322 let mut composite = CompositeSqlBackends::new(default);
323 for (name, db) in databases {
324 let kind = ExternalSqlKind::parse(&db.kind).ok_or_else(|| Error::SqlExternalKind {
325 name: name.clone(),
326 kind: db.kind.clone(),
327 })?;
328 if db.compute.as_deref().is_some_and(|c| !c.is_empty()) {
329 if let Some(var) = db.password_env.as_deref().filter(|v| !v.is_empty()) {
337 let workload = db.compute.as_deref().expect("compute checked above");
338 let password =
339 std::env::var(var).map_err(|_| Error::SqlEnvUnset(var.into()))?;
340 let resolver = Arc::new(crate::managed_sql::DeployEndpointResolver::new(
341 deploy.clone(),
342 boatramp_core::project::DEFAULT_PROJECT,
343 ));
344 let external: Arc<dyn SqlBackend> = Arc::new(
345 boatramp_storage::sql_compute::ComputeResolvedSqlBackend::new(
346 resolver,
347 workload,
348 kind,
349 db.database.clone().unwrap_or_default(),
350 db.user.clone().unwrap_or_default(),
351 password,
352 db.pool_max,
353 db.read_only,
354 timeout(db),
355 ),
356 );
357 composite = composite.with_external(name.clone(), external, db.allow_preview);
358 continue;
359 }
360 let envelope = secrets_envelope
363 .cloned()
364 .ok_or_else(|| Error::SqlManagedNeedsSecrets(name.clone()))?;
365 let resolver = crate::tenant_sql::NodeTenantSqlResolver::new(
366 deploy.clone(),
367 kv.clone(),
368 envelope,
369 db,
370 )
371 .expect("a compute-backed managed binding builds a per-tenant resolver");
372 let site_scoped = resolver.site_scoped();
373 composite = composite.with_per_tenant(
374 name.clone(),
375 Arc::new(resolver),
376 site_scoped,
377 db.allow_preview,
378 );
379 } else {
380 if db.url_env.trim().is_empty() {
383 return Err(Error::SqlExternalUrlEnvMissing(name.clone()));
384 }
385 let url = std::env::var(&db.url_env)
386 .map_err(|_| Error::SqlEnvUnset(db.url_env.clone()))?;
387 let read_url = match &db.read_url_env {
388 Some(var) => {
389 Some(std::env::var(var).map_err(|_| Error::SqlEnvUnset(var.clone()))?)
390 }
391 None => None,
392 };
393 let opts = ExternalSqlOptions::new(url)
394 .with_read_url(read_url)
395 .with_max_connections(db.pool_max)
396 .read_only(db.read_only)
397 .with_connect_timeout(timeout(db));
398 let external: Arc<dyn SqlBackend> =
399 connect(kind, &opts).map_err(|source| Error::SqlExternalConnect {
400 name: name.clone(),
401 source,
402 })?;
403 composite = composite.with_external(name.clone(), external, db.allow_preview);
404 }
405 }
406 Ok(Arc::new(composite))
407 }
408 #[cfg(not(any(feature = "sql-postgres", feature = "sql-mysql")))]
409 {
410 let _ = (deploy, kv, secrets_envelope);
413 let name = databases.keys().next().cloned().unwrap_or_default();
414 Err(Error::SqlExternalUnavailable(name))
415 }
416}
417
418#[cfg(not(feature = "handlers"))]
419#[allow(clippy::too_many_arguments)]
420pub async fn build_handler_runtime(
421 _kv: Arc<dyn KvStore>,
422 _storage: Arc<dyn boatramp_core::Storage>,
423 _data_dir: &Path,
424 _handlers_cfg: Option<&crate::config::HandlersConfig>,
425 _messaging_override: Option<Arc<dyn boatramp_core::messaging::Messaging>>,
426 _max_blob_bytes: u64,
427 _max_component_bytes: u64,
428 _allow_guest_private_egress: bool,
429 _self_egress_addrs: Vec<std::net::SocketAddr>,
430 _guest_egress_extra_roots: Vec<rustls::pki_types::CertificateDer<'static>>,
434 _allow_env_secret_refs: bool,
435 _allow_guest_email: bool,
436 _require_tenancy_declaration: bool,
437 _allow_cross_tenant_db: bool,
438 _deploy: &DeployStore,
439 _secrets_envelope: Option<Arc<dyn KeyEnvelope>>,
440) -> Result<boatramp_server::HandlerRuntime> {
441 Ok(boatramp_server::HandlerRuntime::disabled())
442}
443
444#[cfg(all(test, any(feature = "sql-postgres", feature = "sql-mysql")))]
445mod tests {
446 use super::*;
447 use std::result::Result;
450
451 use async_trait::async_trait;
452 use boatramp_core::envelope::EnvelopeError;
453 use boatramp_core::kv::MemoryKv;
454 use boatramp_core::{ByteStream, GetObject, ObjectMeta, PutMeta, Storage, StorageError};
455
456 struct TestEnvelope;
458 #[async_trait]
459 impl KeyEnvelope for TestEnvelope {
460 async fn wrap(&self, p: &[u8]) -> Result<Vec<u8>, EnvelopeError> {
461 Ok(p.iter().rev().copied().collect())
462 }
463 async fn unwrap(&self, w: &[u8]) -> Result<Vec<u8>, EnvelopeError> {
464 Ok(w.iter().rev().copied().collect())
465 }
466 }
467
468 struct NullStorage;
471 #[async_trait]
472 impl Storage for NullStorage {
473 async fn get(&self, _: &str) -> Result<GetObject, StorageError> {
474 Err(StorageError::NotFound(String::new()))
475 }
476 async fn get_range(
477 &self,
478 _: &str,
479 _: u64,
480 _: Option<u64>,
481 ) -> Result<GetObject, StorageError> {
482 Err(StorageError::NotFound(String::new()))
483 }
484 async fn put(
485 &self,
486 _: &str,
487 _: ByteStream,
488 _: PutMeta,
489 ) -> Result<ObjectMeta, StorageError> {
490 Err(StorageError::unsupported("null"))
491 }
492 async fn head(&self, _: &str) -> Result<ObjectMeta, StorageError> {
493 Err(StorageError::NotFound(String::new()))
494 }
495 async fn delete(&self, _: &str) -> Result<(), StorageError> {
496 Ok(())
497 }
498 async fn list(&self, _: &str) -> Result<Vec<ObjectMeta>, StorageError> {
499 Ok(Vec::new())
500 }
501 }
502
503 fn managed_sql_cfg() -> crate::config::SqlBindingConfig {
505 let mut databases = std::collections::BTreeMap::new();
506 databases.insert(
507 "analytics".to_string(),
508 crate::config::ExternalDatabaseConfig {
509 kind: "postgres".into(),
510 compute: Some("pg".into()),
511 database: Some("analytics".into()),
512 user: Some("app".into()),
513 ..Default::default()
514 },
515 );
516 crate::config::SqlBindingConfig {
517 databases,
518 ..Default::default()
519 }
520 }
521
522 #[tokio::test]
523 async fn managed_sql_fails_closed_without_secrets() {
524 let tmp = tempfile::tempdir().unwrap();
525 let deploy = DeployStore::new(Arc::new(NullStorage), Arc::new(MemoryKv::new()));
526 let kv: Arc<dyn KvStore> = Arc::new(MemoryKv::new());
527 let cfg = managed_sql_cfg();
528 match build_sql_backends(Some(&cfg), tmp.path(), &deploy, &kv, None).await {
530 Err(Error::SqlManagedNeedsSecrets(name)) => assert_eq!(name, "analytics"),
531 Ok(_) => panic!("a managed DB without [secrets] must fail closed, got Ok"),
532 Err(other) => panic!("expected SqlManagedNeedsSecrets, got: {other}"),
533 }
534 }
535
536 #[tokio::test]
537 async fn managed_sql_builds_lazily_and_seals_the_credential() {
538 let tmp = tempfile::tempdir().unwrap();
539 let deploy = DeployStore::new(Arc::new(NullStorage), Arc::new(MemoryKv::new()));
540 let kv: Arc<dyn KvStore> = Arc::new(MemoryKv::new());
541 let envelope: Arc<dyn KeyEnvelope> = Arc::new(TestEnvelope);
542 let cfg = managed_sql_cfg();
543 let backends = build_sql_backends(Some(&cfg), tmp.path(), &deploy, &kv, Some(&envelope))
548 .await
549 .expect("managed sql builds without a live DB (lazy connect)");
550 assert!(
551 kv.get("managed-sql-cred/default/pg")
552 .await
553 .unwrap()
554 .is_none(),
555 "nothing sealed at build — per-tenant credentials are minted on first open"
556 );
557
558 let _ = backends
562 .database("default", "blog", "analytics")
563 .await
564 .unwrap();
565 let sealed = kv
566 .get("managed-sql-cred/default/pg")
567 .await
568 .unwrap()
569 .expect("credential sealed on first resolve under the default project");
570 assert_ne!(sealed.len(), 0);
571 }
572}