1use std::collections::BTreeMap;
17
18use serde::{Deserialize, Serialize};
19
20use crate::config::{ConsumerConfig, DeployConfig, HandlerConfig, HandlerLimits, Overlap};
21use crate::file::FileEntry;
22
23#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
27#[serde(rename_all = "snake_case")]
28pub enum Owner {
29 Site(String),
31 Project(String),
33}
34
35impl std::fmt::Display for Owner {
36 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
37 match self {
38 Self::Site(s) => write!(f, "site:{s}"),
39 Self::Project(p) => write!(f, "project:{p}"),
40 }
41 }
42}
43
44#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, Serialize, Deserialize)]
46#[serde(rename_all = "kebab-case")]
47pub enum Lifecycle {
48 #[default]
50 DeployPinned,
51 Independent,
53}
54
55#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, Serialize, Deserialize)]
59#[serde(rename_all = "snake_case")]
60pub enum Runtime {
61 #[default]
62 Wasm,
63 Microvm,
64 Container,
65}
66
67impl Runtime {
68 pub fn as_str(self) -> &'static str {
70 match self {
71 Self::Wasm => "wasm",
72 Self::Microvm => "microvm",
73 Self::Container => "container",
74 }
75 }
76}
77
78impl std::fmt::Display for Runtime {
79 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
80 f.write_str(self.as_str())
81 }
82}
83
84#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)]
89#[serde(default, deny_unknown_fields)]
90pub struct FunctionConfig {
91 #[serde(skip_serializing_if = "Vec::is_empty")]
93 pub imports: Vec<String>,
94 #[serde(skip_serializing_if = "Option::is_none")]
96 pub limits: Option<HandlerLimits>,
97 #[serde(skip_serializing_if = "BTreeMap::is_empty")]
99 pub env: BTreeMap<String, String>,
100 #[serde(skip_serializing_if = "BTreeMap::is_empty")]
105 pub secrets: BTreeMap<String, String>,
106 pub runtime: Runtime,
108 #[serde(default, skip_serializing_if = "FunctionQuota::is_unset")]
110 pub quota: FunctionQuota,
111 #[serde(default, skip_serializing_if = "Option::is_none")]
113 pub webhook: Option<WebhookConfig>,
114 #[serde(skip_serializing_if = "Vec::is_empty")]
120 pub invoke_targets: Vec<String>,
121 #[serde(
127 default,
128 deserialize_with = "crate::tenancy::de_opt_tenancy",
129 skip_serializing_if = "Option::is_none"
130 )]
131 pub tenancy: Option<crate::tenancy::Tenancy>,
132 #[serde(default, skip_serializing_if = "Option::is_none")]
137 pub token_claims: Option<crate::config::HandlerGraphqlTokenClaims>,
138}
139
140#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, Serialize, Deserialize)]
142#[serde(rename_all = "snake_case")]
143pub enum WebhookAlgorithm {
144 #[default]
146 HmacSha256,
147}
148
149#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
154#[serde(deny_unknown_fields)]
155pub struct WebhookConfig {
156 pub secret_env: String,
158 #[serde(default)]
160 pub algorithm: WebhookAlgorithm,
161 #[serde(default, skip_serializing_if = "Option::is_none")]
164 pub signature_header: Option<String>,
165 #[serde(default, skip_serializing_if = "Option::is_none")]
167 pub max_body_bytes: Option<u64>,
168 #[serde(default, skip_serializing_if = "Option::is_none")]
174 pub publish: Option<String>,
175}
176
177impl WebhookConfig {
178 pub fn header(&self) -> &str {
180 self.signature_header
181 .as_deref()
182 .unwrap_or("x-boatramp-signature")
183 }
184 pub fn body_cap(&self) -> u64 {
186 self.max_body_bytes.unwrap_or(1024 * 1024)
187 }
188}
189
190impl FunctionConfig {
191 fn from_handler(h: &HandlerConfig) -> Self {
192 Self {
193 imports: h.imports.clone(),
194 limits: h.limits.clone(),
195 env: h.env.clone(),
196 secrets: BTreeMap::new(),
200 runtime: Runtime::default(),
201 quota: FunctionQuota::default(),
202 webhook: None,
203 invoke_targets: Vec::new(),
204 tenancy: h.tenancy.clone(),
207 token_claims: h.token_claims.clone(),
208 }
209 }
210 fn from_consumer(c: &ConsumerConfig) -> Self {
211 Self {
212 imports: c.imports.clone(),
213 tenancy: c.tenancy.clone(),
216 token_claims: c.token_claims.clone(),
217 ..Default::default()
218 }
219 }
220}
221
222#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
224#[serde(deny_unknown_fields)]
225pub struct FunctionRef {
226 pub name: String,
228 #[serde(default, skip_serializing_if = "Option::is_none")]
230 pub version: Option<String>,
231}
232
233#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
237#[serde(deny_unknown_fields)]
238pub struct Trigger {
239 pub kind: TriggerKind,
241 #[serde(default, skip_serializing_if = "Option::is_none")]
243 pub target: Option<FunctionRef>,
244}
245
246#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
249#[serde(tag = "type", rename_all = "snake_case")]
250pub enum TriggerKind {
251 Route {
253 #[serde(default, skip_serializing_if = "Option::is_none")]
254 host: Option<String>,
255 path: String,
256 #[serde(default, skip_serializing_if = "Vec::is_empty")]
257 methods: Vec<String>,
258 },
259 Invoke { name: String },
261 Queue {
265 topic: String,
266 #[serde(default, skip_serializing_if = "String::is_empty")]
267 group: String,
268 #[serde(default, skip_serializing_if = "crate::config::is_default_start")]
269 start: crate::config::StartPosition,
270 },
271 Cron {
273 schedule: String,
274 #[serde(default)]
275 overlap: Overlap,
276 },
277 Blob { prefix: String },
279 Webhook { path: String, secret_env: String },
282 Stream {
285 topics: Vec<String>,
286 #[serde(default, skip_serializing_if = "std::ops::Not::not")]
287 websocket: bool,
288 #[serde(default, skip_serializing_if = "Option::is_none")]
289 publish_topic: Option<String>,
290 },
291}
292
293impl std::fmt::Display for Trigger {
294 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
296 match &self.kind {
297 TriggerKind::Route { path, methods, .. } => {
298 let m = if methods.is_empty() {
299 "*".to_string()
300 } else {
301 methods.join(",")
302 };
303 write!(f, "route {m} {path}")
304 }
305 TriggerKind::Invoke { name } => write!(f, "invoke {name}"),
306 TriggerKind::Queue { topic, .. } => write!(f, "queue {topic}"),
307 TriggerKind::Cron { schedule, .. } => write!(f, "cron {schedule}"),
308 TriggerKind::Blob { prefix } => write!(f, "blob {prefix}"),
309 TriggerKind::Webhook { path, .. } => write!(f, "webhook {path}"),
310 TriggerKind::Stream { topics, .. } => write!(f, "stream {}", topics.join(",")),
311 }
312 }
313}
314
315#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
320#[serde(deny_unknown_fields)]
321pub struct FunctionTrigger {
322 pub id: String,
324 pub kind: TriggerKind,
326 #[serde(default, skip_serializing_if = "Option::is_none")]
330 pub last_fired_minute: Option<i64>,
331}
332
333#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
335#[serde(deny_unknown_fields)]
336pub struct Function {
337 pub name: String,
339 pub owner: Owner,
341 pub versions: Vec<FunctionVersion>,
343 pub active: String,
345 #[serde(default, skip_serializing_if = "BTreeMap::is_empty")]
347 pub aliases: BTreeMap<String, String>,
348 pub config: FunctionConfig,
350}
351
352#[derive(Debug, Clone, PartialEq, Eq, thiserror::Error)]
356#[error("no version {id:?} in function {function:?}")]
357pub struct UnknownVersion {
358 pub function: String,
360 pub id: String,
362}
363
364impl Function {
365 pub fn new(
368 name: impl Into<String>,
369 owner: Owner,
370 component_hash: impl Into<String>,
371 config: FunctionConfig,
372 lifecycle: Lifecycle,
373 created: u64,
374 ) -> Self {
375 let hash = component_hash.into();
376 Self {
377 name: name.into(),
378 owner,
379 versions: vec![FunctionVersion {
380 id: hash.clone(),
381 component: hash.clone(),
382 created,
383 lifecycle,
384 }],
385 active: hash,
386 aliases: BTreeMap::new(),
387 config,
388 }
389 }
390
391 pub fn upsert_version(
395 &mut self,
396 component_hash: impl Into<String>,
397 lifecycle: Lifecycle,
398 created: u64,
399 ) -> String {
400 let hash = component_hash.into();
401 if !self.versions.iter().any(|v| v.id == hash) {
402 self.versions.push(FunctionVersion {
403 id: hash.clone(),
404 component: hash.clone(),
405 created,
406 lifecycle,
407 });
408 }
409 self.active = hash.clone();
410 hash
411 }
412
413 pub fn rollback(&mut self, to: &str) -> Result<(), UnknownVersion> {
415 if self.versions.iter().any(|v| v.id == to) {
416 self.active = to.to_string();
417 Ok(())
418 } else {
419 Err(UnknownVersion {
420 function: self.name.clone(),
421 id: to.to_string(),
422 })
423 }
424 }
425
426 pub fn set_alias(&mut self, label: &str, version: &str) -> Result<(), UnknownVersion> {
428 if self.versions.iter().any(|v| v.id == version) {
429 self.aliases.insert(label.to_string(), version.to_string());
430 Ok(())
431 } else {
432 Err(UnknownVersion {
433 function: self.name.clone(),
434 id: version.to_string(),
435 })
436 }
437 }
438
439 pub fn resolve(&self, reference: &str) -> Option<&str> {
443 let id = self
444 .aliases
445 .get(reference)
446 .map(String::as_str)
447 .unwrap_or(reference);
448 self.versions
449 .iter()
450 .find(|v| v.id == id)
451 .map(|v| v.component.as_str())
452 }
453}
454
455#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
457#[serde(deny_unknown_fields)]
458pub struct FunctionVersion {
459 pub id: String,
461 pub component: String,
463 pub created: u64,
465 #[serde(default)]
467 pub lifecycle: Lifecycle,
468}
469
470#[derive(Debug, Clone, PartialEq, Eq)]
475pub struct FunctionSpec {
476 pub name: String,
478 pub component: String,
480 pub config: FunctionConfig,
482 pub lifecycle: Lifecycle,
484}
485
486#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, Serialize, Deserialize)]
490#[serde(rename_all = "snake_case")]
491pub enum InvokeMode {
492 #[default]
494 Sync,
495 Async,
497}
498
499#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, Serialize, Deserialize)]
501#[serde(rename_all = "snake_case")]
502pub enum InvocationStatus {
503 #[default]
505 Queued,
506 Running,
508 Succeeded,
510 Failed,
512}
513
514#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
517#[serde(deny_unknown_fields)]
518pub struct InvocationResult {
519 pub status: u16,
521 #[serde(default, skip_serializing_if = "Option::is_none")]
523 pub content_type: Option<String>,
524 pub body_b64: String,
526}
527
528#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
531#[serde(deny_unknown_fields)]
532pub struct Invocation {
533 pub id: String,
535 pub function: String,
537 pub version: String,
540 pub mode: InvokeMode,
542 pub status: InvocationStatus,
544 #[serde(default, skip_serializing_if = "Option::is_none")]
546 pub idempotency_key: Option<String>,
547 #[serde(default)]
549 pub attempts: u32,
550 #[serde(default, skip_serializing_if = "Option::is_none")]
556 pub lease_expires: Option<u64>,
557 #[serde(default, skip_serializing_if = "Option::is_none")]
559 pub request_b64: Option<String>,
560 #[serde(default, skip_serializing_if = "Option::is_none")]
562 pub request_content_type: Option<String>,
563 #[serde(default, skip_serializing_if = "Option::is_none")]
565 pub result: Option<InvocationResult>,
566 pub created: u64,
568 pub updated: u64,
570}
571
572impl Invocation {
573 pub fn is_terminal(&self) -> bool {
575 matches!(
576 self.status,
577 InvocationStatus::Succeeded | InvocationStatus::Failed
578 )
579 }
580}
581
582#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)]
587#[serde(default, deny_unknown_fields)]
588pub struct FunctionQuota {
589 #[serde(skip_serializing_if = "Option::is_none")]
592 pub max_invocations: Option<u64>,
593 #[serde(skip_serializing_if = "Option::is_none")]
596 pub window_secs: Option<u64>,
597 #[serde(skip_serializing_if = "Option::is_none")]
599 pub max_concurrent: Option<u32>,
600}
601
602impl FunctionQuota {
603 pub fn is_unset(&self) -> bool {
606 self.max_invocations.is_none() && self.max_concurrent.is_none()
607 }
608 pub fn window(&self) -> u64 {
610 self.window_secs.unwrap_or(60)
611 }
612}
613
614#[derive(Debug, Clone, Copy, PartialEq, Eq)]
616pub struct MeteringSample {
617 pub success: bool,
619 pub duration_ms: u64,
622 pub bytes_in: u64,
624 pub bytes_out: u64,
626}
627
628#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)]
632#[serde(default, deny_unknown_fields)]
633pub struct Metering {
634 pub function: String,
636 pub invocations: u64,
638 pub successes: u64,
640 pub failures: u64,
642 pub duration_ms_total: u64,
644 pub bytes_in_total: u64,
646 pub bytes_out_total: u64,
648 pub window_start: u64,
650 pub window_count: u64,
652 pub updated: u64,
654}
655
656impl Metering {
657 pub fn new(function: impl Into<String>) -> Self {
659 Self {
660 function: function.into(),
661 ..Default::default()
662 }
663 }
664
665 pub fn record(&mut self, sample: &MeteringSample, now: u64) {
667 self.invocations += 1;
668 if sample.success {
669 self.successes += 1;
670 } else {
671 self.failures += 1;
672 }
673 self.duration_ms_total = self.duration_ms_total.saturating_add(sample.duration_ms);
674 self.bytes_in_total = self.bytes_in_total.saturating_add(sample.bytes_in);
675 self.bytes_out_total = self.bytes_out_total.saturating_add(sample.bytes_out);
676 self.updated = now;
677 }
678
679 pub fn admit(&mut self, quota: &FunctionQuota, now: u64) -> bool {
684 let Some(max) = quota.max_invocations else {
685 return true;
686 };
687 let window = quota.window();
688 if now.saturating_sub(self.window_start) >= window {
689 self.window_start = now;
690 self.window_count = 0;
691 }
692 if self.window_count >= max {
693 return false;
694 }
695 self.window_count += 1;
696 self.updated = now;
697 true
698 }
699}
700
701pub mod keys {
702 pub fn meta(project: &str, name: &str) -> String {
709 format!("project/{project}/functions/{name}")
710 }
711 pub fn version(project: &str, name: &str, id: &str) -> String {
713 format!("project/{project}/functions/{name}/versions/{id}")
714 }
715 pub fn alias(project: &str, name: &str, label: &str) -> String {
717 format!("project/{project}/functions/{name}/alias/{label}")
718 }
719 pub fn trigger(project: &str, name: &str, id: &str) -> String {
721 format!("project/{project}/functions/{name}/triggers/{id}")
722 }
723 pub fn invocation(project: &str, name: &str, id: &str) -> String {
725 format!("project/{project}/functions/{name}/invocations/{id}")
726 }
727 pub fn invocations_prefix(project: &str, name: &str) -> String {
729 format!("project/{project}/functions/{name}/invocations/")
730 }
731 pub fn idempotency(project: &str, name: &str, key: &str) -> String {
733 format!("project/{project}/functions/{name}/idem/{key}")
734 }
735 pub fn metering(project: &str, name: &str) -> String {
737 format!("project/{project}/metering/{name}")
738 }
739
740 pub fn functions_prefix(project: &str) -> String {
744 format!("project/{project}/functions/")
745 }
746 pub fn triggers_prefix(project: &str, name: &str) -> String {
748 format!("project/{project}/functions/{name}/triggers/")
749 }
750 pub fn metering_prefix(project: &str) -> String {
752 format!("project/{project}/metering/")
753 }
754}
755
756pub fn handler_name(route: &str) -> String {
759 let s = slug(route);
760 if s.is_empty() {
761 "root".to_string()
762 } else {
763 s
764 }
765}
766
767pub fn consumer_name(topic: &str) -> String {
769 format!("consumer-{}", slug(topic))
770}
771
772fn slug(s: &str) -> String {
774 let mut out = String::new();
775 let mut dash = false;
776 for c in s.chars() {
777 if c.is_ascii_alphanumeric() {
778 out.push(c.to_ascii_lowercase());
779 dash = false;
780 } else if !out.is_empty() && !dash {
781 out.push('-');
782 dash = true;
783 }
784 }
785 out.trim_matches('-').to_string()
786}
787
788pub fn desugar(cfg: &DeployConfig) -> (Vec<FunctionSpec>, Vec<Trigger>) {
800 let mut functions = Vec::new();
801 let mut triggers = Vec::new();
802
803 for h in &cfg.handlers {
804 let name = handler_name(&h.route);
805 functions.push(FunctionSpec {
806 name: name.clone(),
807 component: h.component.clone(),
808 config: FunctionConfig::from_handler(h),
809 lifecycle: Lifecycle::DeployPinned,
810 });
811 triggers.push(Trigger {
812 kind: TriggerKind::Route {
813 host: None,
814 path: h.route.clone(),
815 methods: h.methods.clone(),
816 },
817 target: Some(FunctionRef {
818 name,
819 version: None,
820 }),
821 });
822 }
823
824 for c in &cfg.consumers {
825 let name = consumer_name(&c.topic);
826 functions.push(FunctionSpec {
827 name: name.clone(),
828 component: c.component.clone(),
829 config: FunctionConfig::from_consumer(c),
830 lifecycle: Lifecycle::DeployPinned,
831 });
832 triggers.push(Trigger {
833 kind: TriggerKind::Queue {
834 topic: c.topic.clone(),
835 group: c.group.clone(),
836 start: c.start,
837 },
838 target: Some(FunctionRef {
839 name,
840 version: None,
841 }),
842 });
843 }
844
845 for cr in &cfg.crons {
846 let target = cfg
848 .handlers
849 .iter()
850 .find(|h| h.route == cr.route)
851 .map(|h| FunctionRef {
852 name: handler_name(&h.route),
853 version: None,
854 });
855 triggers.push(Trigger {
856 kind: TriggerKind::Cron {
857 schedule: cr.schedule.clone(),
858 overlap: cr.overlap,
859 },
860 target,
861 });
862 }
863
864 for s in &cfg.streams {
865 triggers.push(Trigger {
866 kind: TriggerKind::Stream {
867 topics: s.topics.clone(),
868 websocket: s.websocket,
869 publish_topic: s.publish_topic.clone(),
870 },
871 target: None,
872 });
873 }
874
875 (functions, triggers)
876}
877
878pub fn materialize(
885 specs: &[FunctionSpec],
886 site: &str,
887 files: &BTreeMap<String, FileEntry>,
888 created: u64,
889) -> Vec<Function> {
890 specs
891 .iter()
892 .filter_map(|s| {
893 let hash = files.get(s.component.trim_start_matches('/'))?.hash.clone();
894 Some(Function {
895 name: s.name.clone(),
896 owner: Owner::Site(site.to_string()),
897 versions: vec![FunctionVersion {
898 id: hash.clone(),
899 component: hash.clone(),
900 created,
901 lifecycle: s.lifecycle,
902 }],
903 active: hash,
904 aliases: BTreeMap::new(),
905 config: s.config.clone(),
906 })
907 })
908 .collect()
909}
910
911#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
915pub struct FunctionSummary {
916 pub name: String,
918 pub owner: String,
920 pub runtime: String,
922 pub version: String,
924 pub triggers: Vec<String>,
926}
927
928#[cfg(test)]
929mod tests {
930 use super::*;
931 use crate::config::{ConsumerConfig, CronConfig, HandlerConfig, StreamConfig};
932
933 fn handler(route: &str, component: &str, methods: &[&str], imports: &[&str]) -> HandlerConfig {
934 HandlerConfig {
935 tenancy: None,
936 token_claims: None,
937 route: route.into(),
938 methods: methods
939 .iter()
940 .map(std::string::ToString::to_string)
941 .collect(),
942 component: component.into(),
943 imports: imports
944 .iter()
945 .map(std::string::ToString::to_string)
946 .collect(),
947 streaming: false,
948 limits: None,
949 env: BTreeMap::new(),
950 invoke_targets: Vec::new(),
951 }
952 }
953
954 #[test]
955 fn slugs_and_names() {
956 assert_eq!(handler_name("/api/hello"), "api-hello");
957 assert_eq!(handler_name("/"), "root");
958 assert_eq!(handler_name("/a/b/*"), "a-b");
959 assert_eq!(consumer_name("orders.new"), "consumer-orders-new");
960 }
961
962 #[test]
966 fn desugar_preserves_all_compute_config() {
967 let cfg = DeployConfig {
968 handlers: vec![
969 handler("/api/hello", "hello.wasm", &["GET"], &["kv"]),
970 handler("/api/report", "report.wasm", &[], &[]),
971 ],
972 consumers: vec![ConsumerConfig {
973 tenancy: None,
974 token_claims: None,
975 topic: "orders".into(),
976 component: "orders.wasm".into(),
977 imports: vec!["sql".into()],
978 group: String::new(),
979 start: Default::default(),
980 }],
981 crons: vec![CronConfig {
982 schedule: "0 * * * *".into(),
983 route: "/api/report".into(),
984 overlap: Overlap::Skip,
985 }],
986 streams: vec![StreamConfig {
987 route: "/live".into(),
988 topics: vec!["ticks".into()],
989 websocket: false,
990 publish_topic: None,
991 }],
992 ..Default::default()
993 };
994
995 let (functions, triggers) = desugar(&cfg);
996
997 assert_eq!(functions.len(), 3);
999 let hello = functions.iter().find(|f| f.name == "api-hello").unwrap();
1000 assert_eq!(hello.component, "hello.wasm");
1001 assert_eq!(hello.config.imports, vec!["kv".to_string()]);
1002 assert_eq!(hello.lifecycle, Lifecycle::DeployPinned);
1003 assert_eq!(hello.config.runtime, Runtime::Wasm);
1004 let consumer = functions
1005 .iter()
1006 .find(|f| f.name == "consumer-orders")
1007 .unwrap();
1008 assert_eq!(consumer.component, "orders.wasm");
1009 assert_eq!(consumer.config.imports, vec!["sql".to_string()]);
1010
1011 assert_eq!(triggers.len(), 5);
1013
1014 let route = triggers
1016 .iter()
1017 .find(|t| matches!(&t.kind, TriggerKind::Route { path, .. } if path == "/api/hello"))
1018 .unwrap();
1019 match &route.kind {
1020 TriggerKind::Route { methods, host, .. } => {
1021 assert_eq!(methods, &["GET".to_string()]);
1022 assert!(host.is_none());
1023 }
1024 _ => unreachable!(),
1025 }
1026 assert_eq!(route.target.as_ref().unwrap().name, "api-hello");
1027
1028 let queue = triggers
1030 .iter()
1031 .find(|t| matches!(&t.kind, TriggerKind::Queue { topic, .. } if topic == "orders"))
1032 .unwrap();
1033 assert_eq!(queue.target.as_ref().unwrap().name, "consumer-orders");
1034
1035 let cron = triggers
1037 .iter()
1038 .find(|t| matches!(&t.kind, TriggerKind::Cron { .. }))
1039 .unwrap();
1040 assert_eq!(cron.target.as_ref().unwrap().name, "api-report");
1041
1042 let stream = triggers
1044 .iter()
1045 .find(|t| matches!(&t.kind, TriggerKind::Stream { .. }))
1046 .unwrap();
1047 assert!(stream.target.is_none());
1048 match &stream.kind {
1049 TriggerKind::Stream { topics, .. } => assert_eq!(topics, &["ticks".to_string()]),
1050 _ => unreachable!(),
1051 }
1052 }
1053
1054 #[test]
1055 fn materialize_resolves_paths_to_blob_hashes() {
1056 let cfg = DeployConfig {
1057 handlers: vec![handler("/api/hello", "hello.wasm", &["GET"], &[])],
1058 ..Default::default()
1059 };
1060 let (specs, _) = desugar(&cfg);
1061 let files = BTreeMap::from([(
1062 "hello.wasm".to_string(),
1063 FileEntry {
1064 hash: "sha256:abc".into(),
1065 size: 10,
1066 content_type: None,
1067 variants: BTreeMap::new(),
1068 },
1069 )]);
1070 let funcs = materialize(&specs, "blog", &files, 1_800_000_000);
1071 assert_eq!(funcs.len(), 1);
1072 assert_eq!(funcs[0].name, "api-hello");
1073 assert_eq!(funcs[0].owner, Owner::Site("blog".into()));
1074 assert_eq!(funcs[0].active, "sha256:abc");
1075 assert_eq!(funcs[0].versions[0].component, "sha256:abc");
1076 assert_eq!(funcs[0].versions[0].created, 1_800_000_000);
1077 assert!(materialize(&specs, "blog", &BTreeMap::new(), 0).is_empty());
1079 }
1080
1081 #[test]
1082 fn empty_config_desugars_to_nothing() {
1083 let (functions, triggers) = desugar(&DeployConfig::default());
1084 assert!(functions.is_empty() && triggers.is_empty());
1085 }
1086
1087 #[test]
1088 fn model_serde_round_trips() {
1089 let f = Function {
1090 name: "resize".into(),
1091 owner: Owner::Project("acme".into()),
1092 versions: vec![FunctionVersion {
1093 id: "v1abc".into(),
1094 component: "blob:deadbeef".into(),
1095 created: 1_800_000_000,
1096 lifecycle: Lifecycle::Independent,
1097 }],
1098 active: "v1abc".into(),
1099 aliases: BTreeMap::from([("prod".into(), "v1abc".into())]),
1100 config: FunctionConfig {
1101 imports: vec!["blobstore".into()],
1102 runtime: Runtime::Microvm,
1103 ..Default::default()
1104 },
1105 };
1106 let json = serde_json::to_string(&f).unwrap();
1107 assert_eq!(serde_json::from_str::<Function>(&json).unwrap(), f);
1108
1109 for t in [
1111 Trigger {
1112 kind: TriggerKind::Route {
1113 host: Some("example.com".into()),
1114 path: "/x".into(),
1115 methods: vec!["POST".into()],
1116 },
1117 target: Some(FunctionRef {
1118 name: "resize".into(),
1119 version: None,
1120 }),
1121 },
1122 Trigger {
1123 kind: TriggerKind::Stream {
1124 topics: vec!["t".into()],
1125 websocket: true,
1126 publish_topic: Some("up".into()),
1127 },
1128 target: None,
1129 },
1130 ] {
1131 let j = serde_json::to_string(&t).unwrap();
1132 assert_eq!(serde_json::from_str::<Trigger>(&j).unwrap(), t);
1133 }
1134 }
1135
1136 #[test]
1137 fn versioning_alias_and_rollback() {
1138 let mut f = Function::new(
1139 "resize",
1140 Owner::Project("acme".into()),
1141 "hashA",
1142 FunctionConfig::default(),
1143 Lifecycle::Independent,
1144 1,
1145 );
1146 assert_eq!(f.active, "hashA");
1147 assert_eq!(f.versions.len(), 1);
1148
1149 f.upsert_version("hashB", Lifecycle::Independent, 2);
1151 assert_eq!(f.active, "hashB");
1152 assert_eq!(f.versions.len(), 2);
1153 f.upsert_version("hashA", Lifecycle::Independent, 3);
1155 assert_eq!(f.active, "hashA");
1156 assert_eq!(f.versions.len(), 2);
1157
1158 f.set_alias("prod", "hashB").unwrap();
1160 assert_eq!(f.aliases.get("prod").map(String::as_str), Some("hashB"));
1161 assert!(f.set_alias("prod", "ghost").is_err());
1162
1163 f.rollback("hashB").unwrap();
1165 assert_eq!(f.active, "hashB");
1166 assert!(f.rollback("ghost").is_err());
1167
1168 assert_eq!(f.resolve("hashA"), Some("hashA"));
1171 assert_eq!(f.resolve("prod"), Some("hashB")); assert_eq!(f.resolve("ghost"), None);
1173 }
1174
1175 #[test]
1176 fn invocation_model_round_trips_and_reports_terminal() {
1177 let inv = Invocation {
1178 id: "inv-1".into(),
1179 function: "greeter".into(),
1180 version: "hashA".into(),
1181 mode: InvokeMode::Async,
1182 status: InvocationStatus::Queued,
1183 idempotency_key: Some("k".into()),
1184 attempts: 0,
1185 lease_expires: None,
1186 request_b64: Some("aGk=".into()),
1187 request_content_type: Some("text/plain".into()),
1188 result: None,
1189 created: 1,
1190 updated: 1,
1191 };
1192 assert!(!inv.is_terminal());
1193 let json = serde_json::to_string(&inv).unwrap();
1194 let back: Invocation = serde_json::from_str(&json).unwrap();
1195 assert_eq!(back, inv);
1196 assert!(json.contains("\"mode\":\"async\""));
1198 assert!(json.contains("\"status\":\"queued\""));
1199
1200 let done = Invocation {
1201 status: InvocationStatus::Succeeded,
1202 result: Some(InvocationResult {
1203 status: 200,
1204 content_type: None,
1205 body_b64: "b2s=".into(),
1206 }),
1207 ..inv
1208 };
1209 assert!(done.is_terminal());
1210 }
1211
1212 #[test]
1213 fn metering_records_and_rate_limits() {
1214 let mut m = Metering::new("greeter");
1215 m.record(
1216 &MeteringSample {
1217 success: true,
1218 duration_ms: 5,
1219 bytes_in: 3,
1220 bytes_out: 7,
1221 },
1222 100,
1223 );
1224 m.record(
1225 &MeteringSample {
1226 success: false,
1227 duration_ms: 2,
1228 bytes_in: 0,
1229 bytes_out: 0,
1230 },
1231 101,
1232 );
1233 assert_eq!(m.invocations, 2);
1234 assert_eq!(m.successes, 1);
1235 assert_eq!(m.failures, 1);
1236 assert_eq!(m.duration_ms_total, 7);
1237 assert_eq!(m.bytes_out_total, 7);
1238 assert_eq!(m.updated, 101);
1239
1240 let quota = FunctionQuota {
1242 max_invocations: Some(2),
1243 window_secs: Some(10),
1244 max_concurrent: None,
1245 };
1246 let mut r = Metering::new("greeter");
1247 assert!(r.admit("a, 1000)); assert!(r.admit("a, 1001)); assert!(!r.admit("a, 1002)); assert_eq!(r.window_count, 2);
1251 assert!(r.admit("a, 1011));
1253 assert_eq!(r.window_count, 1);
1254
1255 let unset = FunctionQuota::default();
1257 let mut u = Metering::new("greeter");
1258 assert!(u.admit(&unset, 1));
1259 assert_eq!(u.window_count, 0);
1260 assert!(unset.is_unset());
1261 }
1262
1263 #[test]
1264 fn webhook_config_defaults_and_round_trips() {
1265 let w = WebhookConfig {
1267 secret_env: "HOOK_SECRET".into(),
1268 algorithm: WebhookAlgorithm::HmacSha256,
1269 signature_header: None,
1270 max_body_bytes: None,
1271 publish: None,
1272 };
1273 assert_eq!(w.header(), "x-boatramp-signature");
1274 assert_eq!(w.body_cap(), 1024 * 1024);
1275
1276 let cfg = FunctionConfig {
1278 webhook: Some(w),
1279 ..Default::default()
1280 };
1281 let json = serde_json::to_string(&cfg).unwrap();
1282 assert!(json.contains("\"secret_env\":\"HOOK_SECRET\""));
1283 assert!(json.contains("\"hmac_sha256\""));
1284 let back: FunctionConfig = serde_json::from_str(&json).unwrap();
1285 assert_eq!(back, cfg);
1286
1287 let custom = WebhookConfig {
1289 secret_env: "S".into(),
1290 algorithm: WebhookAlgorithm::HmacSha256,
1291 signature_header: Some("x-hub-signature-256".into()),
1292 max_body_bytes: Some(4096),
1293 publish: None,
1294 };
1295 assert_eq!(custom.header(), "x-hub-signature-256");
1296 assert_eq!(custom.body_cap(), 4096);
1297 }
1298
1299 #[test]
1300 fn secret_refs_round_trip_as_references() {
1301 let cfg = FunctionConfig {
1305 env: BTreeMap::from([("STAGE".into(), "prod".into())]),
1306 secrets: BTreeMap::from([("DB_URL".into(), "PROD_DB_URL".into())]),
1307 ..Default::default()
1308 };
1309 let json = serde_json::to_string(&cfg).unwrap();
1310 assert!(json.contains("\"secrets\":{\"DB_URL\":\"PROD_DB_URL\"}"));
1312 let back: FunctionConfig = serde_json::from_str(&json).unwrap();
1313 assert_eq!(back, cfg);
1314 assert_eq!(
1315 back.secrets.get("DB_URL").map(String::as_str),
1316 Some("PROD_DB_URL")
1317 );
1318
1319 let bare = FunctionConfig::default();
1321 assert!(!serde_json::to_string(&bare).unwrap().contains("secrets"));
1322 }
1323
1324 #[test]
1325 fn keyspace_is_stable() {
1326 assert_eq!(
1327 keys::meta("default", "resize"),
1328 "project/default/functions/resize"
1329 );
1330 assert_eq!(
1331 keys::version("default", "resize", "v1"),
1332 "project/default/functions/resize/versions/v1"
1333 );
1334 assert_eq!(
1335 keys::alias("default", "resize", "prod"),
1336 "project/default/functions/resize/alias/prod"
1337 );
1338 assert_eq!(
1339 keys::trigger("default", "resize", "t1"),
1340 "project/default/functions/resize/triggers/t1"
1341 );
1342 assert_eq!(
1343 keys::invocation("default", "resize", "inv-1"),
1344 "project/default/functions/resize/invocations/inv-1"
1345 );
1346 assert_eq!(
1347 keys::invocations_prefix("default", "resize"),
1348 "project/default/functions/resize/invocations/"
1349 );
1350 assert_eq!(
1351 keys::idempotency("default", "resize", "k-1"),
1352 "project/default/functions/resize/idem/k-1"
1353 );
1354 assert_eq!(
1355 keys::metering("default", "resize"),
1356 "project/default/metering/resize"
1357 );
1358 assert_eq!(
1360 keys::meta("acme", "resize"),
1361 "project/acme/functions/resize"
1362 );
1363 }
1364}