1use super::*;
8
9use boatramp_core::function::FunctionSummary;
10
11#[derive(serde::Deserialize)]
13pub(super) struct FunctionQuery {
14 site: Option<String>,
15}
16
17pub(super) async fn list_functions(
23 State(deploy): State<DeployStore>,
24 Extension(project): axum::extract::Extension<ProjectContext>,
25 axum::extract::Query(query): axum::extract::Query<FunctionQuery>,
26) -> Response {
27 use boatramp_core::function;
28 let sites = match &query.site {
29 Some(s) => vec![s.clone()],
30 None => match deploy.all_sites(project.as_ref()).await {
31 Ok(s) => s,
32 Err(err) => return deploy_error_response(err),
33 },
34 };
35 let mut out: Vec<FunctionSummary> = Vec::new();
36 for site in sites {
37 let manifest = match deploy.current_manifest(project.as_ref(), &site).await {
38 Ok(Some(m)) => m,
39 Ok(None) => continue,
40 Err(err) => return deploy_error_response(err),
41 };
42 let (specs, triggers) = function::desugar(&manifest.config);
43 for f in function::materialize(&specs, &site, &manifest.files, 0) {
44 let trigs = triggers
45 .iter()
46 .filter(|t| t.target.as_ref().map(|r| r.name.as_str()) == Some(f.name.as_str()))
47 .map(std::string::ToString::to_string)
48 .collect();
49 out.push(FunctionSummary {
50 name: format!("{site}/{}", f.name),
51 owner: format!("site:{site}"),
52 runtime: f.config.runtime.as_str().to_string(),
53 version: f.active,
54 triggers: trigs,
55 });
56 }
57 }
58 if query.site.is_none() {
61 match deploy.list_stored_functions(project.as_ref()).await {
62 Ok(stored) => {
63 for f in stored {
64 out.push(FunctionSummary {
65 name: f.name.clone(),
66 owner: f.owner.to_string(),
67 runtime: f.config.runtime.as_str().to_string(),
68 version: f.active,
69 triggers: vec![format!("invoke {}", f.name)],
71 });
72 }
73 }
74 Err(err) => return deploy_error_response(err),
75 }
76 }
77 Json(out).into_response()
78}
79
80#[derive(serde::Deserialize)]
82pub(super) struct FunctionUpsert {
83 pub(super) component: String,
85 #[serde(default)]
87 pub(super) config: boatramp_core::function::FunctionConfig,
88 #[serde(default)]
91 pub(super) lifecycle: boatramp_core::function::Lifecycle,
92}
93
94#[derive(serde::Deserialize, Default)]
96pub(super) struct DeployFunctionQuery {
97 #[serde(default)]
102 register_subgraph: Option<bool>,
103}
104
105#[cfg(feature = "handlers")]
109fn scan_manifest_sections(bytes: &[u8], f: &mut impl FnMut(&[u8])) {
110 use wasmparser::{Parser, Payload};
111 for payload in Parser::new(0).parse_all(bytes) {
112 match payload {
113 Ok(Payload::CustomSection(reader)) if reader.name() == "boatramp:function-manifest" => {
114 f(reader.data());
115 }
116 Ok(Payload::ModuleSection {
117 unchecked_range, ..
118 }) => scan_manifest_sections(&bytes[unchecked_range], f),
119 Ok(_) => {}
120 Err(_) => return,
121 }
122 }
123}
124
125#[cfg(feature = "handlers")]
130fn component_declares_subgraph(component: &[u8]) -> bool {
131 let mut declared = false;
132 scan_manifest_sections(component, &mut |data| {
133 for line in data.split(|&b| b == b'\n') {
134 if let Ok(v) = serde_json::from_slice::<serde_json::Value>(line) {
135 if v.get("subgraph").and_then(serde_json::Value::as_bool) == Some(true) {
136 declared = true;
137 }
138 }
139 }
140 });
141 declared
142}
143
144#[cfg(feature = "handlers")]
150pub(crate) fn component_declares_streaming_route(component: &[u8], route: &str) -> bool {
151 let mut declared = false;
152 scan_manifest_sections(component, &mut |data| {
153 for line in data.split(|&b| b == b'\n') {
154 let Ok(v) = serde_json::from_slice::<serde_json::Value>(line) else {
155 continue;
156 };
157 if v.get("streaming").and_then(serde_json::Value::as_bool) != Some(true) {
158 continue;
159 }
160 let Some(triggers) = v.get("triggers").and_then(serde_json::Value::as_array) else {
162 continue;
163 };
164 for t in triggers {
165 if t.get("on").and_then(serde_json::Value::as_str) != Some("http") {
166 continue;
167 }
168 if let Some(r) = t.get("route").and_then(serde_json::Value::as_str) {
169 let path = r.rsplit(char::is_whitespace).next().unwrap_or(r);
171 if path == route {
172 declared = true;
173 }
174 }
175 }
176 }
177 });
178 declared
179}
180
181#[derive(Debug, Clone, Copy, PartialEq, Eq, serde::Serialize)]
187#[serde(rename_all = "lowercase")]
188pub enum Lifecycle {
189 Stable,
190 Experimental,
191}
192
193impl Lifecycle {
194 #[must_use]
196 pub fn as_str(self) -> &'static str {
197 match self {
198 Self::Stable => "stable",
199 Self::Experimental => "experimental",
200 }
201 }
202}
203
204#[derive(Debug, Clone, serde::Serialize)]
208pub struct CapabilityFeature {
209 pub name: &'static str,
210 pub lifecycle: Lifecycle,
211}
212
213#[cfg(feature = "handlers")]
220pub fn host_capability_features_detailed() -> Vec<CapabilityFeature> {
221 use Lifecycle::{Experimental, Stable};
222 let mut f: Vec<CapabilityFeature> = boatramp_core::config::known_imports()
224 .iter()
225 .map(|&name| CapabilityFeature {
226 name,
227 lifecycle: Stable,
228 })
229 .collect();
230 f.push(CapabilityFeature {
232 name: "sql-json",
233 lifecycle: Stable,
234 });
235 f.push(CapabilityFeature {
238 name: "orm-vector",
239 lifecycle: Experimental,
240 });
241 f.push(CapabilityFeature {
242 name: "streaming",
243 lifecycle: Stable,
244 });
245 if cfg!(feature = "orm-subquery") {
246 f.push(CapabilityFeature {
249 name: "orm-subquery",
250 lifecycle: Experimental,
251 });
252 }
253 f
254}
255
256#[cfg(feature = "handlers")]
259pub fn host_capability_features() -> Vec<&'static str> {
260 host_capability_features_detailed()
261 .into_iter()
262 .map(|c| c.name)
263 .collect()
264}
265
266#[cfg(feature = "handlers")]
270pub fn component_requires(component: &[u8]) -> Vec<String> {
271 let mut reqs = Vec::new();
272 scan_manifest_sections(component, &mut |data| {
273 for line in data.split(|&b| b == b'\n') {
274 if let Ok(v) = serde_json::from_slice::<serde_json::Value>(line) {
275 if let Some(arr) = v.get("requires").and_then(serde_json::Value::as_array) {
276 reqs.extend(arr.iter().filter_map(|r| r.as_str().map(str::to_string)));
277 }
278 }
279 }
280 });
281 reqs.sort();
282 reqs.dedup();
283 reqs
284}
285
286#[cfg(feature = "handlers")]
289pub fn unmet_requires(component: &[u8]) -> Vec<String> {
290 let host = host_capability_features();
291 component_requires(component)
292 .into_iter()
293 .filter(|r| !host.contains(&r.as_str()))
294 .collect()
295}
296
297#[cfg(feature = "handlers")]
306async fn maybe_register_subgraph(
307 deploy: &DeployStore,
308 handlers: &HandlerRuntime,
309 project: boatramp_core::project::ProjectRef<'_>,
310 name: &str,
311 function: &boatramp_core::function::Function,
312 component: &str,
313 register: Option<bool>,
314) -> Result<(), Response> {
315 if register == Some(false) {
316 return Ok(()); }
318 let kv = deploy.kv().as_ref();
319 if !crate::graphql_registry::is_registered_subgraph(kv, project.as_str(), name).await {
320 match crate::handler_dispatch::read_blob_fully(deploy, component).await {
323 Ok(blob) if component_declares_subgraph(&blob) => {}
324 _ => return Ok(()),
325 }
326 }
327 let sdl = match handlers
328 .introspect_subgraph_sdl(deploy, project, function, component)
329 .await
330 {
331 Ok(sdl) => sdl,
332 Err(crate::function_runtime::SubgraphSdlError::Unavailable) => return Ok(()),
334 Err(crate::function_runtime::SubgraphSdlError::NotASubgraph) => {
335 return Err((
336 StatusCode::UNPROCESSABLE_ENTITY,
337 format!(
338 "subgraph `{name}` does not answer `{{ _service {{ sdl }} }}`; deploy with \
339 `?register_subgraph=false` to skip subgraph registration\n"
340 ),
341 )
342 .into_response())
343 }
344 Err(crate::function_runtime::SubgraphSdlError::InvokeFailed(msg)) => {
345 return Err((
346 StatusCode::BAD_GATEWAY,
347 format!("could not introspect subgraph `{name}`: {msg}\n"),
348 )
349 .into_response())
350 }
351 };
352 match crate::graphql_registry::publish(kv, project.as_str(), name, &sdl).await {
353 Ok(_) => Ok(()),
354 Err(crate::graphql_registry::PublishError::Composition(e)) => Err((
355 StatusCode::BAD_REQUEST,
356 format!(
357 "subgraph `{name}` does not compose: {e}\n(deploy with `?register_subgraph=false` \
358 to skip, or unregister a conflicting subgraph first)\n"
359 ),
360 )
361 .into_response()),
362 Err(crate::graphql_registry::PublishError::Store(e)) => Err((
363 StatusCode::INTERNAL_SERVER_ERROR,
364 format!("registry store error: {e}\n"),
365 )
366 .into_response()),
367 }
368}
369
370#[cfg(not(feature = "handlers"))]
372async fn maybe_register_subgraph(
373 _deploy: &DeployStore,
374 _handlers: &HandlerRuntime,
375 _project: boatramp_core::project::ProjectRef<'_>,
376 _name: &str,
377 _function: &boatramp_core::function::Function,
378 _component: &str,
379 _register: Option<bool>,
380) -> Result<(), Response> {
381 Ok(())
382}
383
384pub(super) async fn deploy_function(
389 State(deploy): State<DeployStore>,
390 Extension(project): axum::extract::Extension<ProjectContext>,
391 Extension(handlers): Extension<Arc<HandlerRuntime>>,
392 axum::extract::Query(q): axum::extract::Query<DeployFunctionQuery>,
393 Path(name): Path<String>,
394 Json(body): Json<FunctionUpsert>,
395) -> Response {
396 use boatramp_core::function::{Function, Owner};
397 if let Some(resp) = reject_invalid_name("function", &name) {
398 return resp;
399 }
400 match deploy.has_blob(&body.component).await {
401 Ok(true) => {}
402 Ok(false) => {
403 return (
404 StatusCode::BAD_REQUEST,
405 format!("component blob {} not uploaded\n", body.component),
406 )
407 .into_response()
408 }
409 Err(err) => return deploy_error_response(err),
410 }
411 let now = now_unix();
412 let f = match deploy.get_function(project.as_ref(), &name).await {
413 Ok(Some(mut existing)) => {
414 existing.config = body.config;
415 existing.upsert_version(&body.component, body.lifecycle, now);
416 existing
417 }
418 Ok(None) => Function::new(
421 name.clone(),
422 Owner::Project("default".to_string()),
423 &body.component,
424 body.config,
425 body.lifecycle,
426 now,
427 ),
428 Err(err) => return deploy_error_response(err),
429 };
430 if let Err(resp) = maybe_register_subgraph(
434 &deploy,
435 &handlers,
436 project.as_ref(),
437 &name,
438 &f,
439 &body.component,
440 q.register_subgraph,
441 )
442 .await
443 {
444 return resp;
445 }
446 if let Err(err) = deploy.put_function(project.as_ref(), &f).await {
447 return deploy_error_response(err);
448 }
449 Json(f).into_response()
450}
451
452#[derive(serde::Deserialize)]
454pub(super) struct RollbackBody {
455 pub(super) to: String,
456}
457
458pub(super) async fn rollback_function(
460 State(deploy): State<DeployStore>,
461 Extension(project): axum::extract::Extension<ProjectContext>,
462 Path(name): Path<String>,
463 Json(body): Json<RollbackBody>,
464) -> Response {
465 match deploy.get_function(project.as_ref(), &name).await {
466 Ok(Some(mut f)) => match f.rollback(&body.to) {
467 Ok(()) => {
468 if let Err(err) = deploy.put_function(project.as_ref(), &f).await {
469 return deploy_error_response(err);
470 }
471 Json(f).into_response()
472 }
473 Err(msg) => (StatusCode::BAD_REQUEST, format!("{msg}\n")).into_response(),
474 },
475 Ok(None) => (StatusCode::NOT_FOUND, format!("no function {name:?}\n")).into_response(),
476 Err(err) => deploy_error_response(err),
477 }
478}
479
480#[derive(serde::Deserialize)]
482pub(super) struct AliasBody {
483 pub(super) version: String,
484}
485
486pub(super) async fn alias_function(
488 State(deploy): State<DeployStore>,
489 Extension(project): axum::extract::Extension<ProjectContext>,
490 Path((name, label)): Path<(String, String)>,
491 Json(body): Json<AliasBody>,
492) -> Response {
493 match deploy.get_function(project.as_ref(), &name).await {
494 Ok(Some(mut f)) => match f.set_alias(&label, &body.version) {
495 Ok(()) => {
496 if let Err(err) = deploy.put_function(project.as_ref(), &f).await {
497 return deploy_error_response(err);
498 }
499 Json(f).into_response()
500 }
501 Err(msg) => (StatusCode::BAD_REQUEST, format!("{msg}\n")).into_response(),
502 },
503 Ok(None) => (StatusCode::NOT_FOUND, format!("no function {name:?}\n")).into_response(),
504 Err(err) => deploy_error_response(err),
505 }
506}
507
508pub(super) async fn remove_function(
511 State(deploy): State<DeployStore>,
512 Extension(project): axum::extract::Extension<ProjectContext>,
513 Path(name): Path<String>,
514) -> Response {
515 match deploy.delete_function(project.as_ref(), &name).await {
516 Ok(_) => StatusCode::NO_CONTENT.into_response(),
517 Err(err) => deploy_error_response(err),
518 }
519}
520
521#[cfg(all(test, feature = "handlers"))]
522mod tests {
523 use super::*;
524 use boatramp_core::function::{Function, FunctionConfig, Lifecycle, Owner};
525 use boatramp_core::kv::MemoryKv;
526 use std::sync::Arc;
527
528 #[test]
529 fn host_capability_features_registry_and_requires_filter() {
530 let host = host_capability_features();
531 assert!(host.contains(&"sql"), "base capability token present");
533 assert!(
534 host.contains(&"streaming"),
535 "always-on surface feature present"
536 );
537 assert!(host.contains(&"orm-vector"));
538 assert_eq!(
540 host.contains(&"orm-subquery"),
541 cfg!(feature = "orm-subquery")
542 );
543 let detailed = host_capability_features_detailed();
546 let detailed_names: Vec<&str> = detailed.iter().map(|c| c.name).collect();
547 assert_eq!(
548 detailed_names, host,
549 "detailed names match the flat registry"
550 );
551 let lifecycle = |name: &str| {
552 detailed
553 .iter()
554 .find(|c| c.name == name)
555 .map(|c| c.lifecycle)
556 };
557 assert_eq!(lifecycle("sql"), Some(super::Lifecycle::Stable));
558 assert_eq!(lifecycle("sql-json"), Some(super::Lifecycle::Stable));
559 assert_eq!(
560 lifecycle("orm-vector"),
561 Some(super::Lifecycle::Experimental)
562 );
563 let unmet: Vec<&str> = ["sql", "streaming", "quantum-teleport", "warp-drive"]
565 .into_iter()
566 .filter(|r| !host.contains(r))
567 .collect();
568 assert_eq!(unmet, vec!["quantum-teleport", "warp-drive"]);
569 }
570
571 struct NullStorage;
573 #[async_trait::async_trait]
574 impl boatramp_core::Storage for NullStorage {
575 async fn get(
576 &self,
577 _: &str,
578 ) -> Result<boatramp_core::GetObject, boatramp_core::StorageError> {
579 Err(boatramp_core::StorageError::NotFound(String::new()))
580 }
581 async fn get_range(
582 &self,
583 _: &str,
584 _: u64,
585 _: Option<u64>,
586 ) -> Result<boatramp_core::GetObject, boatramp_core::StorageError> {
587 Err(boatramp_core::StorageError::NotFound(String::new()))
588 }
589 async fn put(
590 &self,
591 _: &str,
592 _: boatramp_core::ByteStream,
593 _: boatramp_core::PutMeta,
594 ) -> Result<boatramp_core::ObjectMeta, boatramp_core::StorageError> {
595 Err(boatramp_core::StorageError::unsupported("null"))
596 }
597 async fn head(
598 &self,
599 _: &str,
600 ) -> Result<boatramp_core::ObjectMeta, boatramp_core::StorageError> {
601 Err(boatramp_core::StorageError::NotFound(String::new()))
602 }
603 async fn delete(&self, _: &str) -> Result<(), boatramp_core::StorageError> {
604 Ok(())
605 }
606 async fn list(
607 &self,
608 _: &str,
609 ) -> Result<Vec<boatramp_core::ObjectMeta>, boatramp_core::StorageError> {
610 Ok(Vec::new())
611 }
612 }
613
614 fn a_function() -> Function {
615 Function::new(
616 "accounts",
617 Owner::Project("default".to_string()),
618 "component-hash",
619 FunctionConfig::default(),
620 Lifecycle::default(),
621 0,
622 )
623 }
624
625 #[tokio::test]
629 async fn refresh_is_a_noop_unless_the_function_is_a_registered_subgraph() {
630 let deploy = DeployStore::new(Arc::new(NullStorage), Arc::new(MemoryKv::new()));
631 let handlers = HandlerRuntime::disabled();
632 let project = boatramp_core::project::ProjectRef::new("default");
633 let f = a_function();
634
635 maybe_register_subgraph(
637 &deploy,
638 &handlers,
639 project,
640 "accounts",
641 &f,
642 "component-hash",
643 None,
644 )
645 .await
646 .expect("an unregistered function deploys freely");
647 assert!(
648 !crate::graphql_registry::is_registered_subgraph(
649 deploy.kv().as_ref(),
650 "default",
651 "accounts"
652 )
653 .await
654 );
655
656 crate::graphql_registry::publish(
658 deploy.kv().as_ref(),
659 "default",
660 "accounts",
661 "type Query { x: Int }",
662 )
663 .await
664 .unwrap();
665 maybe_register_subgraph(
666 &deploy,
667 &handlers,
668 project,
669 "accounts",
670 &f,
671 "component-hash",
672 Some(false),
673 )
674 .await
675 .expect("opt-out never blocks");
676
677 maybe_register_subgraph(
680 &deploy,
681 &handlers,
682 project,
683 "accounts",
684 &f,
685 "component-hash",
686 None,
687 )
688 .await
689 .expect("a node with no engine skips the refresh, it does not block");
690 }
691
692 fn leb128(mut n: usize, out: &mut Vec<u8>) {
696 loop {
697 let mut byte = (n & 0x7f) as u8;
698 n >>= 7;
699 if n != 0 {
700 byte |= 0x80;
701 }
702 out.push(byte);
703 if n == 0 {
704 break;
705 }
706 }
707 }
708
709 fn module_with_manifest(manifest: &[u8]) -> Vec<u8> {
710 let name = b"boatramp:function-manifest";
711 let mut payload = Vec::new();
712 leb128(name.len(), &mut payload);
713 payload.extend_from_slice(name);
714 payload.extend_from_slice(manifest);
715 let mut module = vec![0x00, 0x61, 0x73, 0x6d, 0x01, 0x00, 0x00, 0x00]; module.push(0x00); leb128(payload.len(), &mut module);
718 module.extend_from_slice(&payload);
719 module
720 }
721
722 #[test]
723 fn a_subgraph_marker_in_the_manifest_is_detected() {
724 assert!(component_declares_subgraph(&module_with_manifest(
725 br#"{"name":"schema","triggers":[{"on":"http","route":"POST /graphql"}],"authorize":"public","subgraph":true}"#
726 )));
727 assert!(!component_declares_subgraph(&module_with_manifest(
729 br#"{"name":"orders","authorize":"tenant"}"#
730 )));
731 assert!(!component_declares_subgraph(&[
733 0x00, 0x61, 0x73, 0x6d, 0x01, 0x00, 0x00, 0x00
734 ]));
735 assert!(!component_declares_subgraph(b"not a wasm module"));
737 }
738}