Skip to main content

boatramp_types/
function.rs

1//! The FaaS **function** model — PLAN-faas FA-1.
2//!
3//! A **function** is the compute *artifact*: a versioned WASI component + its
4//! binding/capability config. It is the one primitive the engine runs (decision 1
5//! — "one primitive, two views"). A **handler** is *not* a resource — it is a
6//! function reached by an HTTP **route** trigger; likewise a *consumer* / *cron* is
7//! a function reached by a queue / timer trigger (decision 5). Triggers are their
8//! own objects, and many can point at one function version (decision 2).
9//!
10//! [`desugar`] lowers a site's deploy-scoped `handlers/consumers/crons/streams`
11//! into functions + triggers with **no behavioural change** — the mandatory
12//! non-breaking gate: a site's handlers must run identically before and after.
13//! It is a pure config→shape transform; the content hash / version id of each
14//! function is assigned later, at `sync`, when its component blob is uploaded.
15
16use std::collections::BTreeMap;
17
18use serde::{Deserialize, Serialize};
19
20use crate::config::{ConsumerConfig, DeployConfig, HandlerConfig, HandlerLimits, Overlap};
21use crate::file::FileEntry;
22
23/// A function's owner — a **site** or a **project/tenant**. Drives the KV/blob/sql
24/// binding prefix and the inherited RBAC (a site-scoped function gains no privilege
25/// over its site).
26#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
27#[serde(rename_all = "snake_case")]
28pub enum Owner {
29    /// Owned by a site: `fn/<site>/<name>`, binding prefix + RBAC of the site.
30    Site(String),
31    /// A top-level function owned by a project/tenant: `fn/<name>`.
32    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/// A function version's lifecycle (decision 3: `DeployPinned` is the default).
45#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, Serialize, Deserialize)]
46#[serde(rename_all = "kebab-case")]
47pub enum Lifecycle {
48    /// Versions + rolls back atomically with the owning site's deploy.
49    #[default]
50    DeployPinned,
51    /// Its own version / alias / rollback, independent of any deploy.
52    Independent,
53}
54
55/// The execution substrate (decision 1: a per-function knob; `wasm` is the default
56/// and scales to zero by instantiation; the stronger-isolation substrates are the
57/// compute backends — see PLAN-compute-backends).
58#[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    /// The snake_case wire term (matches the serde `rename_all`).
69    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/// A function's binding/capability config — the `HandlerConfig` capability fields
85/// (imports, resource limits, static env) *minus its trigger* (route/methods, which
86/// become a [`Trigger`]), plus the [`Runtime`] knob. A `HandlerConfig` *is* a
87/// function's config, so the engine has one path.
88#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)]
89#[serde(default, deny_unknown_fields)]
90pub struct FunctionConfig {
91    /// Requested host capabilities (interface names).
92    #[serde(skip_serializing_if = "Vec::is_empty")]
93    pub imports: Vec<String>,
94    /// Optional resource limits (mem / timeout / fuel).
95    #[serde(skip_serializing_if = "Option::is_none")]
96    pub limits: Option<HandlerLimits>,
97    /// Static, non-secret environment.
98    #[serde(skip_serializing_if = "BTreeMap::is_empty")]
99    pub env: BTreeMap<String, String>,
100    /// Execution substrate (default `wasm`).
101    pub runtime: Runtime,
102    /// Usage quota (FA-4). Absent / all-`None` ⇒ unlimited.
103    #[serde(default, skip_serializing_if = "FunctionQuota::is_unset")]
104    pub quota: FunctionQuota,
105    /// Signed inbound-webhook ingress (FA-5). Absent ⇒ no webhook endpoint.
106    #[serde(default, skip_serializing_if = "Option::is_none")]
107    pub webhook: Option<WebhookConfig>,
108    /// Function-to-function invoke allowlist (FI): the target names this function
109    /// may call through the `invoke` capability. Each entry may use `*` wildcards
110    /// (`*` = any sibling, `img-*` = a family, `resize` = one literal). Deny by
111    /// default: empty ⇒ the function cannot invoke anything, even if it imports
112    /// `invoke`. Only consulted when `imports` contains `invoke`.
113    #[serde(skip_serializing_if = "Vec::is_empty")]
114    pub invoke_targets: Vec<String>,
115}
116
117/// The signature scheme a webhook is verified under (FA-5).
118#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, Serialize, Deserialize)]
119#[serde(rename_all = "snake_case")]
120pub enum WebhookAlgorithm {
121    /// `HMAC-SHA256(body, secret)`, hex — the GitHub/Stripe-style scheme.
122    #[default]
123    HmacSha256,
124}
125
126/// Signed inbound-webhook config for a function (FA-5). The verifying secret is a
127/// **reference to a host env var** (never stored plaintext — mirrors site
128/// secrets); the endpoint verifies the request signature over the raw body,
129/// constant-time, **before** the guest runs.
130#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
131#[serde(deny_unknown_fields)]
132pub struct WebhookConfig {
133    /// The host env var holding the shared secret.
134    pub secret_env: String,
135    /// Signature scheme (default HMAC-SHA256).
136    #[serde(default)]
137    pub algorithm: WebhookAlgorithm,
138    /// The header carrying the hex signature (default `x-boatramp-signature`; a
139    /// leading `sha256=` is accepted and stripped).
140    #[serde(default, skip_serializing_if = "Option::is_none")]
141    pub signature_header: Option<String>,
142    /// Max request body accepted before verifying/dispatching (default 1 MiB).
143    #[serde(default, skip_serializing_if = "Option::is_none")]
144    pub max_body_bytes: Option<u64>,
145}
146
147impl WebhookConfig {
148    /// The signature header name (defaulted).
149    pub fn header(&self) -> &str {
150        self.signature_header
151            .as_deref()
152            .unwrap_or("x-boatramp-signature")
153    }
154    /// The body cap (defaulted to 1 MiB).
155    pub fn body_cap(&self) -> u64 {
156        self.max_body_bytes.unwrap_or(1024 * 1024)
157    }
158}
159
160impl FunctionConfig {
161    fn from_handler(h: &HandlerConfig) -> Self {
162        Self {
163            imports: h.imports.clone(),
164            limits: h.limits.clone(),
165            env: h.env.clone(),
166            runtime: Runtime::default(),
167            quota: FunctionQuota::default(),
168            webhook: None,
169            invoke_targets: Vec::new(),
170        }
171    }
172    fn from_consumer(c: &ConsumerConfig) -> Self {
173        Self {
174            imports: c.imports.clone(),
175            ..Default::default()
176        }
177    }
178}
179
180/// A reference to a function (a `None` version = the function's active version).
181#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
182#[serde(deny_unknown_fields)]
183pub struct FunctionRef {
184    /// The function name (site-scoped names are `<site>/<name>`).
185    pub name: String,
186    /// A specific version id, or `None` for the active version.
187    #[serde(default, skip_serializing_if = "Option::is_none")]
188    pub version: Option<String>,
189}
190
191/// A **trigger** — a separate object (decision 2). Many triggers may point at the
192/// same function version (e.g. a route *and* a cron). A `target` of `None` is a
193/// host-native trigger (a stream fan-out has no component).
194#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
195#[serde(deny_unknown_fields)]
196pub struct Trigger {
197    /// What fires the trigger.
198    pub kind: TriggerKind,
199    /// The function it invokes, or `None` for a host-native trigger.
200    #[serde(default, skip_serializing_if = "Option::is_none")]
201    pub target: Option<FunctionRef>,
202}
203
204/// The event that fires a trigger. The role words (route/queue/cron/stream) are the
205/// familiar site-view names; each is just *a way to reach a function*.
206#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
207#[serde(tag = "type", rename_all = "snake_case")]
208pub enum TriggerKind {
209    /// An HTTP route — the "handler" shape. `host` scopes it to a virtualhost.
210    Route {
211        #[serde(default, skip_serializing_if = "Option::is_none")]
212        host: Option<String>,
213        path: String,
214        #[serde(default, skip_serializing_if = "Vec::is_empty")]
215        methods: Vec<String>,
216    },
217    /// A stable invoke name — `/api/functions/<name>` (the FaaS verb, FA-3).
218    Invoke { name: String },
219    /// A message topic — the "consumer" shape.
220    Queue { topic: String },
221    /// A cron schedule — the "cron" shape.
222    Cron {
223        schedule: String,
224        #[serde(default)]
225        overlap: Overlap,
226    },
227    /// An object-storage change under a prefix (FA-5).
228    Blob { prefix: String },
229    /// A signed inbound webhook (FA-5). `secret_env` names the host env var
230    /// holding the verifying secret (never the secret itself).
231    Webhook { path: String, secret_env: String },
232    /// Host-native SSE / WebSocket topic fan-out — the "stream" shape (no component,
233    /// so a stream trigger's `target` is `None`).
234    Stream {
235        topics: Vec<String>,
236        #[serde(default, skip_serializing_if = "std::ops::Not::not")]
237        websocket: bool,
238        #[serde(default, skip_serializing_if = "Option::is_none")]
239        publish_topic: Option<String>,
240    },
241}
242
243impl std::fmt::Display for Trigger {
244    /// A short one-line label for the functions view (`route GET /x`, `queue t`, …).
245    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
246        match &self.kind {
247            TriggerKind::Route { path, methods, .. } => {
248                let m = if methods.is_empty() {
249                    "*".to_string()
250                } else {
251                    methods.join(",")
252                };
253                write!(f, "route {m} {path}")
254            }
255            TriggerKind::Invoke { name } => write!(f, "invoke {name}"),
256            TriggerKind::Queue { topic } => write!(f, "queue {topic}"),
257            TriggerKind::Cron { schedule, .. } => write!(f, "cron {schedule}"),
258            TriggerKind::Blob { prefix } => write!(f, "blob {prefix}"),
259            TriggerKind::Webhook { path, .. } => write!(f, "webhook {path}"),
260            TriggerKind::Stream { topics, .. } => write!(f, "stream {}", topics.join(",")),
261        }
262    }
263}
264
265/// A **stored trigger** bound to a top-level function — the durable form the
266/// server dispatches (FA-3 *scheduled* + FA-5 *event sources*). Its owning
267/// function is the key context, so there is no separate `target`. Keyed under
268/// [`keys::trigger`].
269#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
270#[serde(deny_unknown_fields)]
271pub struct FunctionTrigger {
272    /// Unique id within the function (the key suffix).
273    pub id: String,
274    /// What fires it.
275    pub kind: TriggerKind,
276    /// Cron dedup: the minute-stamp (`hour*60 + minute` within the day, or a
277    /// monotonic per-fire stamp) this trigger last fired at — durable so a fire
278    /// isn't repeated across a restart or (in a cluster) a leader change.
279    #[serde(default, skip_serializing_if = "Option::is_none")]
280    pub last_fired_minute: Option<i64>,
281}
282
283/// A stored, content-addressed function resource (the FA-1/FA-2 keyspace form).
284#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
285#[serde(deny_unknown_fields)]
286pub struct Function {
287    /// The function name (unique within its owner).
288    pub name: String,
289    /// Who owns it (drives binding prefix + RBAC).
290    pub owner: Owner,
291    /// Immutable versions, newest last.
292    pub versions: Vec<FunctionVersion>,
293    /// The active version's id.
294    pub active: String,
295    /// Named aliases → version id (staging/previews; mirrors deploy aliases).
296    #[serde(default, skip_serializing_if = "BTreeMap::is_empty")]
297    pub aliases: BTreeMap<String, String>,
298    /// Binding/capability config.
299    pub config: FunctionConfig,
300}
301
302/// A [`Function`] operation named a version id the function has no record of
303/// (`rollback`/`set_alias`). A typed error so a caller can distinguish it (e.g.
304/// map it to a 404) instead of string-matching a message.
305#[derive(Debug, Clone, PartialEq, Eq, thiserror::Error)]
306#[error("no version {id:?} in function {function:?}")]
307pub struct UnknownVersion {
308    /// The function the operation was on.
309    pub function: String,
310    /// The version id that was not found.
311    pub id: String,
312}
313
314impl Function {
315    /// A new top-level function with a single active version (the component blob).
316    /// The version id **is** the component's content hash (content-addressed).
317    pub fn new(
318        name: impl Into<String>,
319        owner: Owner,
320        component_hash: impl Into<String>,
321        config: FunctionConfig,
322        lifecycle: Lifecycle,
323        created: u64,
324    ) -> Self {
325        let hash = component_hash.into();
326        Self {
327            name: name.into(),
328            owner,
329            versions: vec![FunctionVersion {
330                id: hash.clone(),
331                component: hash.clone(),
332                created,
333                lifecycle,
334            }],
335            active: hash,
336            aliases: BTreeMap::new(),
337            config,
338        }
339    }
340
341    /// Add a version for `component_hash` (id = the hash) and make it active. If a
342    /// version with that hash already exists it is just re-activated (idempotent).
343    /// Returns the (active) version id.
344    pub fn upsert_version(
345        &mut self,
346        component_hash: impl Into<String>,
347        lifecycle: Lifecycle,
348        created: u64,
349    ) -> String {
350        let hash = component_hash.into();
351        if !self.versions.iter().any(|v| v.id == hash) {
352            self.versions.push(FunctionVersion {
353                id: hash.clone(),
354                component: hash.clone(),
355                created,
356                lifecycle,
357            });
358        }
359        self.active = hash.clone();
360        hash
361    }
362
363    /// Point `active` at an existing version id. `Err` if the version is unknown.
364    pub fn rollback(&mut self, to: &str) -> Result<(), UnknownVersion> {
365        if self.versions.iter().any(|v| v.id == to) {
366            self.active = to.to_string();
367            Ok(())
368        } else {
369            Err(UnknownVersion {
370                function: self.name.clone(),
371                id: to.to_string(),
372            })
373        }
374    }
375
376    /// Set `label` → an existing version id. `Err` if the version is unknown.
377    pub fn set_alias(&mut self, label: &str, version: &str) -> Result<(), UnknownVersion> {
378        if self.versions.iter().any(|v| v.id == version) {
379            self.aliases.insert(label.to_string(), version.to_string());
380            Ok(())
381        } else {
382            Err(UnknownVersion {
383                function: self.name.clone(),
384                id: version.to_string(),
385            })
386        }
387    }
388
389    /// Resolve a **reference** — an alias label or a version id — to the component
390    /// blob hash that backs it, if known. An alias label is resolved first, then a
391    /// version id (so a label named like an id still wins as a label).
392    pub fn resolve(&self, reference: &str) -> Option<&str> {
393        let id = self
394            .aliases
395            .get(reference)
396            .map(String::as_str)
397            .unwrap_or(reference);
398        self.versions
399            .iter()
400            .find(|v| v.id == id)
401            .map(|v| v.component.as_str())
402    }
403}
404
405/// One immutable version of a function.
406#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
407#[serde(deny_unknown_fields)]
408pub struct FunctionVersion {
409    /// Immutable content-hash id.
410    pub id: String,
411    /// The component blob hash.
412    pub component: String,
413    /// Unix creation time.
414    pub created: u64,
415    /// This version's lifecycle.
416    #[serde(default)]
417    pub lifecycle: Lifecycle,
418}
419
420/// The desugared shape of a function derived from a site's `DeployConfig`. The
421/// `component` is still a **path** within the deploy — its content hash / version
422/// id is assigned at `sync`, when the blob is uploaded (so this stays a pure
423/// config→shape transform, testable without a blob store).
424#[derive(Debug, Clone, PartialEq, Eq)]
425pub struct FunctionSpec {
426    /// Site-scoped function name (see [`handler_name`] / [`consumer_name`]).
427    pub name: String,
428    /// The component path within the deploy.
429    pub component: String,
430    /// Binding/capability config.
431    pub config: FunctionConfig,
432    /// Version lifecycle (deploy-scoped functions are `DeployPinned`).
433    pub lifecycle: Lifecycle,
434}
435
436/// How an invocation is delivered (FA-3). `Sync` runs inline and returns the
437/// function's response; `Async` durably enqueues the call, returns `202` with an
438/// id, and a drain worker runs it later (retried, then dead-lettered).
439#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, Serialize, Deserialize)]
440#[serde(rename_all = "snake_case")]
441pub enum InvokeMode {
442    /// Run inline; the caller blocks on the function's response.
443    #[default]
444    Sync,
445    /// Durably enqueue; the caller gets an id to poll.
446    Async,
447}
448
449/// The lifecycle of a durable (async) invocation.
450#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, Serialize, Deserialize)]
451#[serde(rename_all = "snake_case")]
452pub enum InvocationStatus {
453    /// Enqueued, not yet claimed by a drain worker.
454    #[default]
455    Queued,
456    /// Claimed and executing.
457    Running,
458    /// Completed (the function returned a response — any HTTP status).
459    Succeeded,
460    /// Exhausted its attempts and was dead-lettered.
461    Failed,
462}
463
464/// The captured response of a completed invocation (sync idempotency replay +
465/// async poll both read this). The body is base64 so the record is plain JSON.
466#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
467#[serde(deny_unknown_fields)]
468pub struct InvocationResult {
469    /// The HTTP status the function returned.
470    pub status: u16,
471    /// The response content type, if any.
472    #[serde(default, skip_serializing_if = "Option::is_none")]
473    pub content_type: Option<String>,
474    /// The response body, base64 (standard, no padding stripped).
475    pub body_b64: String,
476}
477
478/// A durable invocation record — the unit of the async queue and the receipt an
479/// idempotency key replays. Keyed under [`keys::invocation`].
480#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
481#[serde(deny_unknown_fields)]
482pub struct Invocation {
483    /// Opaque invocation id (also the queue key suffix).
484    pub id: String,
485    /// The function invoked.
486    pub function: String,
487    /// The function version that ran (or will run) — pinned at enqueue so a
488    /// later deploy can't silently change an in-flight async call.
489    pub version: String,
490    /// Delivery mode.
491    pub mode: InvokeMode,
492    /// Current status.
493    pub status: InvocationStatus,
494    /// The idempotency key that created it, if any.
495    #[serde(default, skip_serializing_if = "Option::is_none")]
496    pub idempotency_key: Option<String>,
497    /// Delivery attempts so far (async).
498    #[serde(default)]
499    pub attempts: u32,
500    /// The request body the function receives, base64.
501    #[serde(default, skip_serializing_if = "Option::is_none")]
502    pub request_b64: Option<String>,
503    /// The request content type forwarded to the function.
504    #[serde(default, skip_serializing_if = "Option::is_none")]
505    pub request_content_type: Option<String>,
506    /// The captured result once complete.
507    #[serde(default, skip_serializing_if = "Option::is_none")]
508    pub result: Option<InvocationResult>,
509    /// Unix create time.
510    pub created: u64,
511    /// Unix last-update time.
512    pub updated: u64,
513}
514
515impl Invocation {
516    /// Whether the invocation reached a terminal state.
517    pub fn is_terminal(&self) -> bool {
518        matches!(
519            self.status,
520            InvocationStatus::Succeeded | InvocationStatus::Failed
521        )
522    }
523}
524
525/// Per-function usage quota (FA-4) — `require`-style knobs enforced host-side,
526/// **fail-closed** (over the limit ⇒ `429`). Accounting, not billing: this bounds
527/// abuse, it does not charge. All limits are per node today (a cluster-wide token
528/// bucket is future work); `None` on a field means that dimension is unlimited.
529#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)]
530#[serde(default, deny_unknown_fields)]
531pub struct FunctionQuota {
532    /// Max invocations admitted within [`window_secs`](Self::window_secs) (a
533    /// fixed window: the counter resets when the window rolls over).
534    #[serde(skip_serializing_if = "Option::is_none")]
535    pub max_invocations: Option<u64>,
536    /// The rate-limit window length in seconds (defaults to 60 when a
537    /// `max_invocations` cap is set).
538    #[serde(skip_serializing_if = "Option::is_none")]
539    pub window_secs: Option<u64>,
540    /// Max concurrent in-flight invocations for this function (per node).
541    #[serde(skip_serializing_if = "Option::is_none")]
542    pub max_concurrent: Option<u32>,
543}
544
545impl FunctionQuota {
546    /// Whether any dimension is set (an all-`None` quota is a no-op the caller can
547    /// skip entirely).
548    pub fn is_unset(&self) -> bool {
549        self.max_invocations.is_none() && self.max_concurrent.is_none()
550    }
551    /// The effective rate-limit window (defaults to 60s).
552    pub fn window(&self) -> u64 {
553        self.window_secs.unwrap_or(60)
554    }
555}
556
557/// One invocation's measured cost, folded into a [`Metering`] aggregate.
558#[derive(Debug, Clone, Copy, PartialEq, Eq)]
559pub struct MeteringSample {
560    /// Whether the invocation succeeded (the guest produced a response).
561    pub success: bool,
562    /// Wall-clock duration in milliseconds (the server-side CPU-time proxy; true
563    /// fuel accounting needs the engine to surface post-completion cost).
564    pub duration_ms: u64,
565    /// Request bytes delivered to the function.
566    pub bytes_in: u64,
567    /// Response bytes the function produced.
568    pub bytes_out: u64,
569}
570
571/// Host-side usage aggregate for one function (FA-4), tenant-isolated under
572/// [`keys::metering`]. It also carries the fixed-window rate-limit counter so
573/// metering + quota share a single read-modify-write.
574#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)]
575#[serde(default, deny_unknown_fields)]
576pub struct Metering {
577    /// The function these counters belong to.
578    pub function: String,
579    /// Total invocations metered (sync + drained async).
580    pub invocations: u64,
581    /// Invocations whose guest produced a response.
582    pub successes: u64,
583    /// Invocations that failed to deliver (engine error / dead-lettered).
584    pub failures: u64,
585    /// Summed wall-clock duration, milliseconds.
586    pub duration_ms_total: u64,
587    /// Summed request bytes.
588    pub bytes_in_total: u64,
589    /// Summed response bytes.
590    pub bytes_out_total: u64,
591    /// The current rate-limit window's start (unix seconds).
592    pub window_start: u64,
593    /// Invocations admitted in the current window.
594    pub window_count: u64,
595    /// Last update (unix seconds).
596    pub updated: u64,
597}
598
599impl Metering {
600    /// A fresh aggregate for `function`.
601    pub fn new(function: impl Into<String>) -> Self {
602        Self {
603            function: function.into(),
604            ..Default::default()
605        }
606    }
607
608    /// Fold one invocation's cost into the usage counters.
609    pub fn record(&mut self, sample: &MeteringSample, now: u64) {
610        self.invocations += 1;
611        if sample.success {
612            self.successes += 1;
613        } else {
614            self.failures += 1;
615        }
616        self.duration_ms_total = self.duration_ms_total.saturating_add(sample.duration_ms);
617        self.bytes_in_total = self.bytes_in_total.saturating_add(sample.bytes_in);
618        self.bytes_out_total = self.bytes_out_total.saturating_add(sample.bytes_out);
619        self.updated = now;
620    }
621
622    /// Admit one invocation against the fixed-window rate limit, rolling the
623    /// window over when it has elapsed. Returns `true` if admitted (and records
624    /// the admission), `false` if the window is already at `max` (⇒ the caller
625    /// fails closed with `429`). An unset cap always admits.
626    pub fn admit(&mut self, quota: &FunctionQuota, now: u64) -> bool {
627        let Some(max) = quota.max_invocations else {
628            return true;
629        };
630        let window = quota.window();
631        if now.saturating_sub(self.window_start) >= window {
632            self.window_start = now;
633            self.window_count = 0;
634        }
635        if self.window_count >= max {
636            return false;
637        }
638        self.window_count += 1;
639        self.updated = now;
640        true
641    }
642}
643
644pub mod keys {
645    //! KV keyspace for a function (mirrors the deploy/alias immutability model), all
646    //! **project-scoped** under `project/<proj>/…` (0.2.0). `project` is passed as a
647    //! bare `&str` (this crate is wasm-clean and has no `ProjectRef`); callers in
648    //! `boatramp-core` pass `ProjectRef::as_str()`.
649
650    /// Function metadata.
651    pub fn meta(project: &str, name: &str) -> String {
652        format!("project/{project}/functions/{name}")
653    }
654    /// An immutable version.
655    pub fn version(project: &str, name: &str, id: &str) -> String {
656        format!("project/{project}/functions/{name}/versions/{id}")
657    }
658    /// A named alias → version id.
659    pub fn alias(project: &str, name: &str, label: &str) -> String {
660        format!("project/{project}/functions/{name}/alias/{label}")
661    }
662    /// A trigger bound to the function.
663    pub fn trigger(project: &str, name: &str, id: &str) -> String {
664        format!("project/{project}/functions/{name}/triggers/{id}")
665    }
666    /// A durable invocation record.
667    pub fn invocation(project: &str, name: &str, id: &str) -> String {
668        format!("project/{project}/functions/{name}/invocations/{id}")
669    }
670    /// The prefix under which all of a function's invocations live (queue scan).
671    pub fn invocations_prefix(project: &str, name: &str) -> String {
672        format!("project/{project}/functions/{name}/invocations/")
673    }
674    /// An idempotency key → invocation id pointer.
675    pub fn idempotency(project: &str, name: &str, key: &str) -> String {
676        format!("project/{project}/functions/{name}/idem/{key}")
677    }
678    /// The function's usage-metering aggregate (FA-4).
679    pub fn metering(project: &str, name: &str) -> String {
680        format!("project/{project}/metering/{name}")
681    }
682
683    /// The prefix listing every function record in a project (`meta` keys). A
684    /// function record's key has no further `/` after its name, so a lister filters
685    /// on that to skip the per-function sub-records.
686    pub fn functions_prefix(project: &str) -> String {
687        format!("project/{project}/functions/")
688    }
689    /// The prefix listing a function's triggers.
690    pub fn triggers_prefix(project: &str, name: &str) -> String {
691        format!("project/{project}/functions/{name}/triggers/")
692    }
693    /// The prefix listing every metering aggregate in a project.
694    pub fn metering_prefix(project: &str) -> String {
695        format!("project/{project}/metering/")
696    }
697}
698
699/// The site-scoped function name for an HTTP route handler — a slug of the route
700/// (`/api/hello` → `api-hello`, `/` → `root`).
701pub fn handler_name(route: &str) -> String {
702    let s = slug(route);
703    if s.is_empty() {
704        "root".to_string()
705    } else {
706        s
707    }
708}
709
710/// The site-scoped function name for a topic consumer (`orders` → `consumer-orders`).
711pub fn consumer_name(topic: &str) -> String {
712    format!("consumer-{}", slug(topic))
713}
714
715/// Lower-case alphanumeric slug, non-alnum runs collapsed to single `-`, trimmed.
716fn slug(s: &str) -> String {
717    let mut out = String::new();
718    let mut dash = false;
719    for c in s.chars() {
720        if c.is_ascii_alphanumeric() {
721            out.push(c.to_ascii_lowercase());
722            dash = false;
723        } else if !out.is_empty() && !dash {
724            out.push('-');
725            dash = true;
726        }
727    }
728    out.trim_matches('-').to_string()
729}
730
731/// Lower a site's deploy-scoped compute config into **functions + triggers**
732/// (decision 2), preserving behaviour exactly (the non-breaking gate):
733///
734/// - each `handler` → a `DeployPinned` [`FunctionSpec`] + a `Route` [`Trigger`];
735/// - each `consumer` → a [`FunctionSpec`] + a `Queue` trigger;
736/// - each `cron` → a `Cron` trigger onto the **function serving its route** (a
737///   second trigger on one function — N triggers → one function);
738/// - each `stream` → a host-native `Stream` trigger (`target: None`, no component).
739///
740/// Pure: no I/O, no hashing. The site handler surface (`routing.handlers`) is
741/// unchanged — this is the internal lowering that FA-1..FA-3 build on.
742pub fn desugar(cfg: &DeployConfig) -> (Vec<FunctionSpec>, Vec<Trigger>) {
743    let mut functions = Vec::new();
744    let mut triggers = Vec::new();
745
746    for h in &cfg.handlers {
747        let name = handler_name(&h.route);
748        functions.push(FunctionSpec {
749            name: name.clone(),
750            component: h.component.clone(),
751            config: FunctionConfig::from_handler(h),
752            lifecycle: Lifecycle::DeployPinned,
753        });
754        triggers.push(Trigger {
755            kind: TriggerKind::Route {
756                host: None,
757                path: h.route.clone(),
758                methods: h.methods.clone(),
759            },
760            target: Some(FunctionRef {
761                name,
762                version: None,
763            }),
764        });
765    }
766
767    for c in &cfg.consumers {
768        let name = consumer_name(&c.topic);
769        functions.push(FunctionSpec {
770            name: name.clone(),
771            component: c.component.clone(),
772            config: FunctionConfig::from_consumer(c),
773            lifecycle: Lifecycle::DeployPinned,
774        });
775        triggers.push(Trigger {
776            kind: TriggerKind::Queue {
777                topic: c.topic.clone(),
778            },
779            target: Some(FunctionRef {
780                name,
781                version: None,
782            }),
783        });
784    }
785
786    for cr in &cfg.crons {
787        // A cron fires an existing handler-function, addressed by its route.
788        let target = cfg
789            .handlers
790            .iter()
791            .find(|h| h.route == cr.route)
792            .map(|h| FunctionRef {
793                name: handler_name(&h.route),
794                version: None,
795            });
796        triggers.push(Trigger {
797            kind: TriggerKind::Cron {
798                schedule: cr.schedule.clone(),
799                overlap: cr.overlap,
800            },
801            target,
802        });
803    }
804
805    for s in &cfg.streams {
806        triggers.push(Trigger {
807            kind: TriggerKind::Stream {
808                topics: s.topics.clone(),
809                websocket: s.websocket,
810                publish_topic: s.publish_topic.clone(),
811            },
812            target: None,
813        });
814    }
815
816    (functions, triggers)
817}
818
819/// Materialize desugared specs into stored [`Function`]s for a site, resolving each
820/// component **path** to its blob hash via the deploy's file map (the blob hash is
821/// the content-addressed version id). `created` is the deploy's activation time.
822/// Specs whose component isn't in the file map are dropped (a validated deploy
823/// always has them). This is the derived, read-only view of a site's functions
824/// (FA-1); independently-stored top-level functions come with FA-2.
825pub fn materialize(
826    specs: &[FunctionSpec],
827    site: &str,
828    files: &BTreeMap<String, FileEntry>,
829    created: u64,
830) -> Vec<Function> {
831    specs
832        .iter()
833        .filter_map(|s| {
834            let hash = files.get(s.component.trim_start_matches('/'))?.hash.clone();
835            Some(Function {
836                name: s.name.clone(),
837                owner: Owner::Site(site.to_string()),
838                versions: vec![FunctionVersion {
839                    id: hash.clone(),
840                    component: hash.clone(),
841                    created,
842                    lifecycle: s.lifecycle,
843                }],
844                active: hash,
845                aliases: BTreeMap::new(),
846                config: s.config.clone(),
847            })
848        })
849        .collect()
850}
851
852/// One entry in the `GET /api/functions` view: a function plus its resolved
853/// active version and the triggers that reach it. The read-only projection the
854/// server returns and the CLI/console render.
855#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
856pub struct FunctionSummary {
857    /// Function name (`<site>/<name>` for site-scoped; bare for top-level).
858    pub name: String,
859    /// Owner (`site:<site>` or `project:<project>`).
860    pub owner: String,
861    /// Execution substrate.
862    pub runtime: String,
863    /// Active version id (the component blob hash).
864    pub version: String,
865    /// Rendered triggers that reach this function.
866    pub triggers: Vec<String>,
867}
868
869#[cfg(test)]
870mod tests {
871    use super::*;
872    use crate::config::{ConsumerConfig, CronConfig, HandlerConfig, StreamConfig};
873
874    fn handler(route: &str, component: &str, methods: &[&str], imports: &[&str]) -> HandlerConfig {
875        HandlerConfig {
876            route: route.into(),
877            methods: methods
878                .iter()
879                .map(std::string::ToString::to_string)
880                .collect(),
881            component: component.into(),
882            imports: imports
883                .iter()
884                .map(std::string::ToString::to_string)
885                .collect(),
886            limits: None,
887            env: BTreeMap::new(),
888            invoke_targets: Vec::new(),
889        }
890    }
891
892    #[test]
893    fn slugs_and_names() {
894        assert_eq!(handler_name("/api/hello"), "api-hello");
895        assert_eq!(handler_name("/"), "root");
896        assert_eq!(handler_name("/a/b/*"), "a-b");
897        assert_eq!(consumer_name("orders.new"), "consumer-orders-new");
898    }
899
900    /// The mandatory **non-breaking gate**: desugaring preserves every handler,
901    /// consumer, cron, and stream's fields exactly — the same component, imports,
902    /// routes/methods, topics, and the cron→handler binding.
903    #[test]
904    fn desugar_preserves_all_compute_config() {
905        let cfg = DeployConfig {
906            handlers: vec![
907                handler("/api/hello", "hello.wasm", &["GET"], &["kv"]),
908                handler("/api/report", "report.wasm", &[], &[]),
909            ],
910            consumers: vec![ConsumerConfig {
911                topic: "orders".into(),
912                component: "orders.wasm".into(),
913                imports: vec!["sql".into()],
914            }],
915            crons: vec![CronConfig {
916                schedule: "0 * * * *".into(),
917                route: "/api/report".into(),
918                overlap: Overlap::Skip,
919            }],
920            streams: vec![StreamConfig {
921                route: "/live".into(),
922                topics: vec!["ticks".into()],
923                websocket: false,
924                publish_topic: None,
925            }],
926            ..Default::default()
927        };
928
929        let (functions, triggers) = desugar(&cfg);
930
931        // Two handlers + one consumer → three functions.
932        assert_eq!(functions.len(), 3);
933        let hello = functions.iter().find(|f| f.name == "api-hello").unwrap();
934        assert_eq!(hello.component, "hello.wasm");
935        assert_eq!(hello.config.imports, vec!["kv".to_string()]);
936        assert_eq!(hello.lifecycle, Lifecycle::DeployPinned);
937        assert_eq!(hello.config.runtime, Runtime::Wasm);
938        let consumer = functions
939            .iter()
940            .find(|f| f.name == "consumer-orders")
941            .unwrap();
942        assert_eq!(consumer.component, "orders.wasm");
943        assert_eq!(consumer.config.imports, vec!["sql".to_string()]);
944
945        // Triggers: 2 routes + 1 queue + 1 cron + 1 stream = 5.
946        assert_eq!(triggers.len(), 5);
947
948        // The route trigger carries the exact path + methods and targets its function.
949        let route = triggers
950            .iter()
951            .find(|t| matches!(&t.kind, TriggerKind::Route { path, .. } if path == "/api/hello"))
952            .unwrap();
953        match &route.kind {
954            TriggerKind::Route { methods, host, .. } => {
955                assert_eq!(methods, &["GET".to_string()]);
956                assert!(host.is_none());
957            }
958            _ => unreachable!(),
959        }
960        assert_eq!(route.target.as_ref().unwrap().name, "api-hello");
961
962        // The queue trigger.
963        let queue = triggers
964            .iter()
965            .find(|t| matches!(&t.kind, TriggerKind::Queue { topic } if topic == "orders"))
966            .unwrap();
967        assert_eq!(queue.target.as_ref().unwrap().name, "consumer-orders");
968
969        // The cron is a SECOND trigger on the /api/report function (N triggers → 1 fn).
970        let cron = triggers
971            .iter()
972            .find(|t| matches!(&t.kind, TriggerKind::Cron { .. }))
973            .unwrap();
974        assert_eq!(cron.target.as_ref().unwrap().name, "api-report");
975
976        // The stream is host-native: a trigger with no function target.
977        let stream = triggers
978            .iter()
979            .find(|t| matches!(&t.kind, TriggerKind::Stream { .. }))
980            .unwrap();
981        assert!(stream.target.is_none());
982        match &stream.kind {
983            TriggerKind::Stream { topics, .. } => assert_eq!(topics, &["ticks".to_string()]),
984            _ => unreachable!(),
985        }
986    }
987
988    #[test]
989    fn materialize_resolves_paths_to_blob_hashes() {
990        let cfg = DeployConfig {
991            handlers: vec![handler("/api/hello", "hello.wasm", &["GET"], &[])],
992            ..Default::default()
993        };
994        let (specs, _) = desugar(&cfg);
995        let files = BTreeMap::from([(
996            "hello.wasm".to_string(),
997            FileEntry {
998                hash: "sha256:abc".into(),
999                size: 10,
1000                content_type: None,
1001                variants: BTreeMap::new(),
1002            },
1003        )]);
1004        let funcs = materialize(&specs, "blog", &files, 1_800_000_000);
1005        assert_eq!(funcs.len(), 1);
1006        assert_eq!(funcs[0].name, "api-hello");
1007        assert_eq!(funcs[0].owner, Owner::Site("blog".into()));
1008        assert_eq!(funcs[0].active, "sha256:abc");
1009        assert_eq!(funcs[0].versions[0].component, "sha256:abc");
1010        assert_eq!(funcs[0].versions[0].created, 1_800_000_000);
1011        // A spec whose component blob is absent is dropped (no phantom function).
1012        assert!(materialize(&specs, "blog", &BTreeMap::new(), 0).is_empty());
1013    }
1014
1015    #[test]
1016    fn empty_config_desugars_to_nothing() {
1017        let (functions, triggers) = desugar(&DeployConfig::default());
1018        assert!(functions.is_empty() && triggers.is_empty());
1019    }
1020
1021    #[test]
1022    fn model_serde_round_trips() {
1023        let f = Function {
1024            name: "resize".into(),
1025            owner: Owner::Project("acme".into()),
1026            versions: vec![FunctionVersion {
1027                id: "v1abc".into(),
1028                component: "blob:deadbeef".into(),
1029                created: 1_800_000_000,
1030                lifecycle: Lifecycle::Independent,
1031            }],
1032            active: "v1abc".into(),
1033            aliases: BTreeMap::from([("prod".into(), "v1abc".into())]),
1034            config: FunctionConfig {
1035                imports: vec!["blobstore".into()],
1036                runtime: Runtime::Microvm,
1037                ..Default::default()
1038            },
1039        };
1040        let json = serde_json::to_string(&f).unwrap();
1041        assert_eq!(serde_json::from_str::<Function>(&json).unwrap(), f);
1042
1043        // A trigger with a data-carrying kind + host-native (no target) both round-trip.
1044        for t in [
1045            Trigger {
1046                kind: TriggerKind::Route {
1047                    host: Some("example.com".into()),
1048                    path: "/x".into(),
1049                    methods: vec!["POST".into()],
1050                },
1051                target: Some(FunctionRef {
1052                    name: "resize".into(),
1053                    version: None,
1054                }),
1055            },
1056            Trigger {
1057                kind: TriggerKind::Stream {
1058                    topics: vec!["t".into()],
1059                    websocket: true,
1060                    publish_topic: Some("up".into()),
1061                },
1062                target: None,
1063            },
1064        ] {
1065            let j = serde_json::to_string(&t).unwrap();
1066            assert_eq!(serde_json::from_str::<Trigger>(&j).unwrap(), t);
1067        }
1068    }
1069
1070    #[test]
1071    fn versioning_alias_and_rollback() {
1072        let mut f = Function::new(
1073            "resize",
1074            Owner::Project("acme".into()),
1075            "hashA",
1076            FunctionConfig::default(),
1077            Lifecycle::Independent,
1078            1,
1079        );
1080        assert_eq!(f.active, "hashA");
1081        assert_eq!(f.versions.len(), 1);
1082
1083        // A new component → a new active version.
1084        f.upsert_version("hashB", Lifecycle::Independent, 2);
1085        assert_eq!(f.active, "hashB");
1086        assert_eq!(f.versions.len(), 2);
1087        // Re-deploying the same hash is idempotent (re-activates, no dup version).
1088        f.upsert_version("hashA", Lifecycle::Independent, 3);
1089        assert_eq!(f.active, "hashA");
1090        assert_eq!(f.versions.len(), 2);
1091
1092        // Alias to a known version; unknown is rejected.
1093        f.set_alias("prod", "hashB").unwrap();
1094        assert_eq!(f.aliases.get("prod").map(String::as_str), Some("hashB"));
1095        assert!(f.set_alias("prod", "ghost").is_err());
1096
1097        // Rollback to a known version; unknown is rejected.
1098        f.rollback("hashB").unwrap();
1099        assert_eq!(f.active, "hashB");
1100        assert!(f.rollback("ghost").is_err());
1101
1102        // `resolve` maps a version id OR an alias label to the component hash;
1103        // an unknown reference resolves to nothing.
1104        assert_eq!(f.resolve("hashA"), Some("hashA"));
1105        assert_eq!(f.resolve("prod"), Some("hashB")); // alias → version → component
1106        assert_eq!(f.resolve("ghost"), None);
1107    }
1108
1109    #[test]
1110    fn invocation_model_round_trips_and_reports_terminal() {
1111        let inv = Invocation {
1112            id: "inv-1".into(),
1113            function: "greeter".into(),
1114            version: "hashA".into(),
1115            mode: InvokeMode::Async,
1116            status: InvocationStatus::Queued,
1117            idempotency_key: Some("k".into()),
1118            attempts: 0,
1119            request_b64: Some("aGk=".into()),
1120            request_content_type: Some("text/plain".into()),
1121            result: None,
1122            created: 1,
1123            updated: 1,
1124        };
1125        assert!(!inv.is_terminal());
1126        let json = serde_json::to_string(&inv).unwrap();
1127        let back: Invocation = serde_json::from_str(&json).unwrap();
1128        assert_eq!(back, inv);
1129        // Enum wire forms are snake_case + stable.
1130        assert!(json.contains("\"mode\":\"async\""));
1131        assert!(json.contains("\"status\":\"queued\""));
1132
1133        let done = Invocation {
1134            status: InvocationStatus::Succeeded,
1135            result: Some(InvocationResult {
1136                status: 200,
1137                content_type: None,
1138                body_b64: "b2s=".into(),
1139            }),
1140            ..inv
1141        };
1142        assert!(done.is_terminal());
1143    }
1144
1145    #[test]
1146    fn metering_records_and_rate_limits() {
1147        let mut m = Metering::new("greeter");
1148        m.record(
1149            &MeteringSample {
1150                success: true,
1151                duration_ms: 5,
1152                bytes_in: 3,
1153                bytes_out: 7,
1154            },
1155            100,
1156        );
1157        m.record(
1158            &MeteringSample {
1159                success: false,
1160                duration_ms: 2,
1161                bytes_in: 0,
1162                bytes_out: 0,
1163            },
1164            101,
1165        );
1166        assert_eq!(m.invocations, 2);
1167        assert_eq!(m.successes, 1);
1168        assert_eq!(m.failures, 1);
1169        assert_eq!(m.duration_ms_total, 7);
1170        assert_eq!(m.bytes_out_total, 7);
1171        assert_eq!(m.updated, 101);
1172
1173        // Fixed-window rate limit: 2 per 10s window.
1174        let quota = FunctionQuota {
1175            max_invocations: Some(2),
1176            window_secs: Some(10),
1177            max_concurrent: None,
1178        };
1179        let mut r = Metering::new("greeter");
1180        assert!(r.admit(&quota, 1000)); // 1st in window
1181        assert!(r.admit(&quota, 1001)); // 2nd
1182        assert!(!r.admit(&quota, 1002)); // 3rd → rejected
1183        assert_eq!(r.window_count, 2);
1184        // The window rolls over after `window_secs`, resetting the counter.
1185        assert!(r.admit(&quota, 1011));
1186        assert_eq!(r.window_count, 1);
1187
1188        // An unset cap always admits and never touches the counter.
1189        let unset = FunctionQuota::default();
1190        let mut u = Metering::new("greeter");
1191        assert!(u.admit(&unset, 1));
1192        assert_eq!(u.window_count, 0);
1193        assert!(unset.is_unset());
1194    }
1195
1196    #[test]
1197    fn webhook_config_defaults_and_round_trips() {
1198        // Defaults: header + body cap.
1199        let w = WebhookConfig {
1200            secret_env: "HOOK_SECRET".into(),
1201            algorithm: WebhookAlgorithm::HmacSha256,
1202            signature_header: None,
1203            max_body_bytes: None,
1204        };
1205        assert_eq!(w.header(), "x-boatramp-signature");
1206        assert_eq!(w.body_cap(), 1024 * 1024);
1207
1208        // A config carrying a webhook round-trips and the secret is an env *ref*.
1209        let cfg = FunctionConfig {
1210            webhook: Some(w),
1211            ..Default::default()
1212        };
1213        let json = serde_json::to_string(&cfg).unwrap();
1214        assert!(json.contains("\"secret_env\":\"HOOK_SECRET\""));
1215        assert!(json.contains("\"hmac_sha256\""));
1216        let back: FunctionConfig = serde_json::from_str(&json).unwrap();
1217        assert_eq!(back, cfg);
1218
1219        // Custom header + cap are honoured.
1220        let custom = WebhookConfig {
1221            secret_env: "S".into(),
1222            algorithm: WebhookAlgorithm::HmacSha256,
1223            signature_header: Some("x-hub-signature-256".into()),
1224            max_body_bytes: Some(4096),
1225        };
1226        assert_eq!(custom.header(), "x-hub-signature-256");
1227        assert_eq!(custom.body_cap(), 4096);
1228    }
1229
1230    #[test]
1231    fn keyspace_is_stable() {
1232        assert_eq!(
1233            keys::meta("default", "resize"),
1234            "project/default/functions/resize"
1235        );
1236        assert_eq!(
1237            keys::version("default", "resize", "v1"),
1238            "project/default/functions/resize/versions/v1"
1239        );
1240        assert_eq!(
1241            keys::alias("default", "resize", "prod"),
1242            "project/default/functions/resize/alias/prod"
1243        );
1244        assert_eq!(
1245            keys::trigger("default", "resize", "t1"),
1246            "project/default/functions/resize/triggers/t1"
1247        );
1248        assert_eq!(
1249            keys::invocation("default", "resize", "inv-1"),
1250            "project/default/functions/resize/invocations/inv-1"
1251        );
1252        assert_eq!(
1253            keys::invocations_prefix("default", "resize"),
1254            "project/default/functions/resize/invocations/"
1255        );
1256        assert_eq!(
1257            keys::idempotency("default", "resize", "k-1"),
1258            "project/default/functions/resize/idem/k-1"
1259        );
1260        assert_eq!(
1261            keys::metering("default", "resize"),
1262            "project/default/metering/resize"
1263        );
1264        // A project other than the default keys under its own segment.
1265        assert_eq!(
1266            keys::meta("acme", "resize"),
1267            "project/acme/functions/resize"
1268        );
1269    }
1270}