iii_sdk/
protocol.rs

1use iii_helpers::http::HttpInvocationConfig;
2use schemars::JsonSchema;
3use serde::{Deserialize, Serialize};
4use serde_json::Value;
5use uuid::Uuid;
6
7/// [`Message::RegistrationRejected`] code: another live worker already holds
8/// this `(namespace, worker_name)`. The engine closes the connection; the SDK
9/// must stop and not reconnect. Mirrors the engine constant of the same name.
10pub const WORKER_NAMESPACE_CONFLICT: &str = "WORKER_NAMESPACE_CONFLICT";
11
12/// [`Message::RegistrationRejected`] code: another live worker in this
13/// namespace already exports this function id. Only that one registration is
14/// refused; the connection stays open and the worker keeps serving its other
15/// functions. Mirrors the engine constant of the same name.
16pub const FUNCTION_NAMESPACE_CONFLICT: &str = "FUNCTION_NAMESPACE_CONFLICT";
17
18/// Routing action for [`TriggerRequest`]. Determines how the engine handles
19/// the invocation.
20///
21/// - `Enqueue`: Routes through a named queue for async processing.
22/// - `Void`: Fire-and-forget, no response.
23#[derive(Debug, Clone, Serialize, Deserialize, JsonSchema)]
24#[serde(tag = "type", rename_all = "lowercase")]
25pub enum TriggerAction {
26    /// Routes the invocation through a named queue.
27    Enqueue { queue: String },
28    /// Fire-and-forget routing.
29    Void,
30}
31
32/// Request object for `trigger()`.
33///
34/// ```rust
35/// # use iii_sdk::protocol::{TriggerRequest, TriggerAction};
36/// # use serde_json::json;
37/// // Simple call
38/// TriggerRequest {
39///     function_id: "my::function".to_string(),
40///     payload: json!({ "key": "value" }),
41///     action: None,
42///     timeout_ms: None,
43/// };
44///
45/// // With action
46/// TriggerRequest {
47///     function_id: "my::function".to_string(),
48///     payload: json!({}),
49///     action: Some(TriggerAction::Enqueue { queue: "payments".to_string() }),
50///     timeout_ms: None,
51/// };
52///
53/// // With metadata
54/// TriggerRequest {
55///     function_id: "my::function".to_string(),
56///     payload: json!({}),
57///     action: None,
58///     timeout_ms: None,
59/// }
60/// .metadata(json!({ "tenant": "acme" }));
61/// ```
62#[derive(Debug, Clone)]
63pub struct TriggerRequest {
64    /// ID of the function to invoke.
65    pub function_id: String,
66    /// Input data passed to the function.
67    pub payload: Value,
68    /// Sets how the trigger is routed. `None` for a synchronous request/response.
69    /// Set a routing scheme otherwise (e.g. `TriggerAction::Enqueue { .. }`, `TriggerAction::Void`).
70    pub action: Option<TriggerAction>,
71    /// Override the default invocation timeout, in milliseconds.
72    pub timeout_ms: Option<u64>,
73}
74
75impl TriggerRequest {
76    /// Attach per-invocation metadata without adding a required field to
77    /// [`TriggerRequest`] struct literals.
78    pub fn metadata(self, metadata: Value) -> TriggerRequestWithMetadata {
79        TriggerRequestWithMetadata {
80            request: self,
81            metadata: Some(metadata),
82            namespace: None,
83        }
84    }
85
86    /// Target a specific namespace for this invocation without adding a
87    /// required field to [`TriggerRequest`] struct literals. Serializes into
88    /// [`Message::InvokeFunction`]'s `namespace`.
89    ///
90    /// Leaving it unset does not mean the engine's default: the call inherits
91    /// this worker's namespace. Say `default` to reach the engine's from a
92    /// namespaced worker.
93    pub fn namespace(self, namespace: impl Into<String>) -> TriggerRequestWithMetadata {
94        TriggerRequestWithMetadata {
95            request: self,
96            metadata: None,
97            namespace: Some(namespace.into()),
98        }
99    }
100}
101
102/// Trigger request plus optional per-invocation metadata and target namespace.
103#[derive(Debug, Clone)]
104pub struct TriggerRequestWithMetadata {
105    pub(crate) request: TriggerRequest,
106    pub(crate) metadata: Option<Value>,
107    pub(crate) namespace: Option<String>,
108}
109
110impl TriggerRequestWithMetadata {
111    /// Attach per-invocation metadata.
112    pub fn metadata(mut self, metadata: Value) -> Self {
113        self.metadata = Some(metadata);
114        self
115    }
116
117    /// Target a specific namespace for this invocation.
118    pub fn namespace(mut self, namespace: impl Into<String>) -> Self {
119        self.namespace = Some(namespace.into());
120        self
121    }
122}
123
124impl<T> From<T> for TriggerRequestWithMetadata
125where
126    T: Into<TriggerRequest>,
127{
128    fn from(request: T) -> Self {
129        Self {
130            request: request.into(),
131            metadata: None,
132            namespace: None,
133        }
134    }
135}
136
137#[derive(Debug, Clone, Serialize, Deserialize)]
138#[serde(tag = "type", rename_all = "lowercase")]
139pub enum Message {
140    RegisterTriggerType {
141        id: String,
142        description: String,
143        #[serde(skip_serializing_if = "Option::is_none")]
144        trigger_request_format: Option<Value>,
145        #[serde(skip_serializing_if = "Option::is_none")]
146        call_request_format: Option<Value>,
147        /// Namespace this provider serves. Absent means the connection's own.
148        #[serde(default, skip_serializing_if = "Option::is_none")]
149        namespace: Option<String>,
150    },
151    RegisterTrigger {
152        id: String,
153        trigger_type: String,
154        function_id: String,
155        config: Value,
156        #[serde(skip_serializing_if = "Option::is_none")]
157        metadata: Option<Value>,
158        /// Namespace the trigger's target function resolves in.
159        ///
160        /// The engine reads an absent field as its default namespace. The SDK
161        /// does not leave it absent: `register_trigger` fills it from this
162        /// worker's namespace, because the function a trigger names is one this
163        /// worker registered, and that landed in the worker's namespace. Name
164        /// another namespace, `default` included, to bind elsewhere.
165        #[serde(default, skip_serializing_if = "Option::is_none")]
166        namespace: Option<String>,
167        /// Namespace to find the provider in. Absent asks the engine to resolve
168        /// it: this connection's namespace first, then the engine's own.
169        #[serde(default, skip_serializing_if = "Option::is_none")]
170        trigger_namespace: Option<String>,
171    },
172    TriggerRegistrationResult {
173        id: String,
174        trigger_type: String,
175        function_id: String,
176        #[serde(skip_serializing_if = "Option::is_none")]
177        error: Option<ErrorBody>,
178    },
179    UnregisterTrigger {
180        id: String,
181        trigger_type: String,
182    },
183    UnregisterTriggerType {
184        id: String,
185    },
186    RegisterFunction {
187        id: String,
188        #[serde(skip_serializing_if = "Option::is_none")]
189        description: Option<String>,
190        #[serde(skip_serializing_if = "Option::is_none")]
191        request_format: Option<Value>,
192        #[serde(skip_serializing_if = "Option::is_none")]
193        response_format: Option<Value>,
194        #[serde(skip_serializing_if = "Option::is_none")]
195        metadata: Option<Value>,
196        #[serde(skip_serializing_if = "Option::is_none")]
197        invocation: Option<HttpInvocationConfig>,
198    },
199    UnregisterFunction {
200        id: String,
201    },
202    InvokeFunction {
203        invocation_id: Option<Uuid>,
204        function_id: String,
205        data: Value,
206        #[serde(skip_serializing_if = "Option::is_none")]
207        traceparent: Option<String>,
208        #[serde(skip_serializing_if = "Option::is_none")]
209        baggage: Option<String>,
210        #[serde(skip_serializing_if = "Option::is_none")]
211        action: Option<TriggerAction>,
212        /// Per-invocation metadata sidecar, surfaced to the handler as a
213        /// distinct argument alongside `data`. Optional and additive
214        /// for wire compatibility with engines that don't send it.
215        #[serde(default, skip_serializing_if = "Option::is_none")]
216        metadata: Option<Value>,
217        /// Target namespace for routing. Optional and additive, so older peers
218        /// that don't send it stay wire-compatible, and the engine reads an
219        /// absent field as its default namespace.
220        ///
221        /// A call made through this SDK does not arrive absent: `trigger` fills
222        /// it from this worker's namespace unless the request names one.
223        #[serde(default, skip_serializing_if = "Option::is_none")]
224        namespace: Option<String>,
225    },
226    InvocationResult {
227        invocation_id: Uuid,
228        function_id: String,
229        #[serde(skip_serializing_if = "Option::is_none")]
230        result: Option<Value>,
231        #[serde(skip_serializing_if = "Option::is_none")]
232        error: Option<ErrorBody>,
233        #[serde(skip_serializing_if = "Option::is_none")]
234        traceparent: Option<String>,
235        #[serde(skip_serializing_if = "Option::is_none")]
236        baggage: Option<String>,
237    },
238    Ping,
239    Pong,
240    /// Sent to the engine as the first message of a reconnect, before the
241    /// registration replay: `previous_worker_id` and `reattach_token` are
242    /// the values the engine assigned via `WorkerRegistered` on the previous
243    /// connection. The engine retires that connection so the replay lands on
244    /// a clean slate; the token is required because worker ids alone are
245    /// publicly discoverable.
246    Reattach {
247        previous_worker_id: String,
248        #[serde(default, skip_serializing_if = "Option::is_none")]
249        reattach_token: Option<String>,
250    },
251    WorkerRegistered {
252        worker_id: String,
253        /// Secret to present in `Reattach` on reconnect; absent on older
254        /// engines.
255        #[serde(default, skip_serializing_if = "Option::is_none")]
256        reattach_token: Option<String>,
257    },
258    /// Pushed by the engine when a registration collides with a live worker in
259    /// the same namespace. The `code` distinguishes the two cases:
260    /// [`WORKER_NAMESPACE_CONFLICT`] is fatal (the engine closes the connection;
261    /// the SDK stops and does not reconnect), while [`FUNCTION_NAMESPACE_CONFLICT`]
262    /// refuses a single function id and keeps the connection open.
263    RegistrationRejected {
264        code: String,
265        namespace: String,
266        #[serde(default, skip_serializing_if = "Option::is_none")]
267        worker_name: Option<String>,
268        #[serde(default, skip_serializing_if = "Option::is_none")]
269        function_id: Option<String>,
270        owner_worker_id: String,
271    },
272}
273
274#[derive(Debug, Clone, Serialize, Deserialize)]
275pub struct RegisterTriggerTypeMessage {
276    /// Unique identifier for the trigger type (e.g. `state`, `durable:subscriber`).
277    pub id: String,
278    /// Human-readable description of what this trigger type does.
279    pub description: String,
280    #[serde(skip_serializing_if = "Option::is_none")]
281    pub trigger_request_format: Option<Value>,
282    #[serde(skip_serializing_if = "Option::is_none")]
283    pub call_request_format: Option<Value>,
284    /// Namespace this provider serves. `None` lets the engine use the
285    /// connection's own, which is what a worker providing a trigger type for
286    /// its own project wants.
287    #[serde(default, skip_serializing_if = "Option::is_none")]
288    pub namespace: Option<String>,
289}
290
291impl RegisterTriggerTypeMessage {
292    pub fn to_message(&self) -> Message {
293        Message::RegisterTriggerType {
294            id: self.id.clone(),
295            description: self.description.clone(),
296            trigger_request_format: self.trigger_request_format.clone(),
297            call_request_format: self.call_request_format.clone(),
298            namespace: self.namespace.clone(),
299        }
300    }
301}
302
303/// Input for [`IIIClient::register_trigger`](crate::IIIClient::register_trigger).
304/// The `id` is auto-generated internally.
305///
306/// Build it with [`RegisterTriggerInput::new`], or with
307/// [`IIITrigger`](crate::builtin_triggers::IIITrigger) for a type the engine
308/// provides. A struct literal has to name every field, so each field added here
309/// breaks it -- `namespace` did that once and `trigger_namespace` did it again,
310/// which is why the constructors exist.
311///
312/// The two are different questions. `namespace` is where the target function
313/// resolves; `trigger_namespace` is where the trigger type's provider is found,
314/// and `None` there asks the engine to take this worker's namespace first and
315/// the engine's own second.
316#[derive(Debug, Clone, Serialize, Deserialize)]
317pub struct RegisterTriggerInput {
318    /// Identifier of the registered trigger type this trigger uses (e.g. `storage::object-created`, `http`).
319    pub trigger_type: String,
320    /// ID of the function this trigger invokes when it fires.
321    pub function_id: String,
322    /// Trigger-type-specific configuration, matching the shape the trigger type expects.
323    pub config: Value,
324    /// Arbitrary user-specifiable metadata supplied to the triggered handler function on every invocation.
325    #[serde(skip_serializing_if = "Option::is_none")]
326    pub metadata: Option<Value>,
327    /// Namespace the trigger's target function resolves in. `None` inherits
328    /// this worker's namespace; name another namespace, including `default`,
329    /// to bind the trigger elsewhere.
330    #[serde(default, skip_serializing_if = "Option::is_none")]
331    pub namespace: Option<String>,
332    /// Namespace to find the trigger type's provider in. `None` asks the
333    /// engine to resolve it: this worker's namespace first, the engine's own
334    /// second. Naming one is strict.
335    #[serde(default, skip_serializing_if = "Option::is_none")]
336    pub trigger_namespace: Option<String>,
337}
338
339impl RegisterTriggerInput {
340    /// A trigger of `trigger_type`, bound to `function_id`, configured by
341    /// `config`.
342    ///
343    /// Use this for a trigger type nothing built in provides -- one this worker
344    /// or a neighbour registered. For the engine's own types, prefer
345    /// [`IIITrigger`](crate::builtin_triggers::IIITrigger), which knows the id
346    /// and the config shape:
347    ///
348    /// ```no_run
349    /// # use iii_sdk::protocol::RegisterTriggerInput;
350    /// # use serde_json::json;
351    /// RegisterTriggerInput::new("kick", "orders::sync", json!({ "every": "1m" }))
352    ///     .in_namespace("billing")
353    ///     .in_trigger_namespace("shop-a");
354    /// ```
355    ///
356    /// Everything else defaults to "not named", which is what the great
357    /// majority of registrations want. Reaching for a struct literal instead
358    /// means every field added here becomes a breaking change for the caller,
359    /// which is what this exists to stop.
360    pub fn new(
361        trigger_type: impl Into<String>,
362        function_id: impl Into<String>,
363        config: Value,
364    ) -> Self {
365        Self {
366            trigger_type: trigger_type.into(),
367            function_id: function_id.into(),
368            config,
369            metadata: None,
370            namespace: None,
371            trigger_namespace: None,
372        }
373    }
374
375    /// Resolve this trigger's target function in `namespace` instead of this
376    /// worker's.
377    pub fn in_namespace(mut self, namespace: impl Into<String>) -> Self {
378        self.namespace = Some(namespace.into());
379        self
380    }
381
382    /// Find the trigger type's provider in `namespace`, and only there.
383    ///
384    /// A different question from [`in_namespace`](Self::in_namespace): that one
385    /// locates the target function, this one locates the provider that fires
386    /// it. Left unset, the engine resolves it -- this worker's namespace first,
387    /// then its own -- which is what carries a worker that has not been
388    /// migrated onto the engine's providers without saying anything.
389    pub fn in_trigger_namespace(mut self, namespace: impl Into<String>) -> Self {
390        self.trigger_namespace = Some(namespace.into());
391        self
392    }
393
394    /// Deliver `metadata` to the handler alongside every payload this trigger
395    /// fires.
396    pub fn with_metadata(mut self, metadata: Value) -> Self {
397        self.metadata = Some(metadata);
398        self
399    }
400}
401
402#[derive(Debug, Clone, Serialize, Deserialize)]
403pub struct RegisterTriggerMessage {
404    pub id: String,
405    pub trigger_type: String,
406    pub function_id: String,
407    pub config: Value,
408    #[serde(skip_serializing_if = "Option::is_none")]
409    pub metadata: Option<Value>,
410    #[serde(default, skip_serializing_if = "Option::is_none")]
411    pub namespace: Option<String>,
412    #[serde(default, skip_serializing_if = "Option::is_none")]
413    pub trigger_namespace: Option<String>,
414}
415
416impl RegisterTriggerMessage {
417    pub fn to_message(&self) -> Message {
418        Message::RegisterTrigger {
419            id: self.id.clone(),
420            trigger_type: self.trigger_type.clone(),
421            function_id: self.function_id.clone(),
422            config: self.config.clone(),
423            metadata: self.metadata.clone(),
424            namespace: self.namespace.clone(),
425            trigger_namespace: self.trigger_namespace.clone(),
426        }
427    }
428}
429
430#[derive(Debug, Clone, Serialize, Deserialize)]
431pub struct UnregisterTriggerMessage {
432    pub id: String,
433    pub trigger_type: String,
434}
435
436impl UnregisterTriggerMessage {
437    pub fn to_message(&self) -> Message {
438        Message::UnregisterTrigger {
439            id: self.id.clone(),
440            trigger_type: self.trigger_type.clone(),
441        }
442    }
443}
444
445#[derive(Debug, Clone, Serialize, Deserialize)]
446pub struct UnregisterTriggerTypeMessage {
447    pub id: String,
448}
449
450impl UnregisterTriggerTypeMessage {
451    pub fn to_message(&self) -> Message {
452        Message::UnregisterTriggerType {
453            id: self.id.clone(),
454        }
455    }
456}
457
458#[derive(Debug, Clone, Serialize, Deserialize)]
459pub struct RegisterFunctionMessage {
460    pub id: String,
461    #[serde(skip_serializing_if = "Option::is_none")]
462    pub description: Option<String>,
463    #[serde(skip_serializing_if = "Option::is_none")]
464    pub request_format: Option<Value>,
465    #[serde(skip_serializing_if = "Option::is_none")]
466    pub response_format: Option<Value>,
467    #[serde(skip_serializing_if = "Option::is_none")]
468    pub metadata: Option<Value>,
469    #[serde(skip_serializing_if = "Option::is_none")]
470    pub invocation: Option<HttpInvocationConfig>,
471}
472
473impl RegisterFunctionMessage {
474    pub fn with_id(name: String) -> Self {
475        RegisterFunctionMessage {
476            id: name,
477            description: None,
478            request_format: None,
479            response_format: None,
480            metadata: None,
481            invocation: None,
482        }
483    }
484    pub fn with_description(mut self, description: String) -> Self {
485        self.description = Some(description);
486        self
487    }
488    pub fn to_message(&self) -> Message {
489        Message::RegisterFunction {
490            id: self.id.clone(),
491            description: self.description.clone(),
492            request_format: self.request_format.clone(),
493            response_format: self.response_format.clone(),
494            metadata: self.metadata.clone(),
495            invocation: self.invocation.clone(),
496        }
497    }
498}
499
500#[derive(Debug, Clone, Serialize, Deserialize)]
501pub struct FunctionMessage {
502    pub function_id: String,
503    #[serde(skip_serializing_if = "Option::is_none")]
504    pub description: Option<String>,
505    #[serde(skip_serializing_if = "Option::is_none")]
506    pub request_format: Option<Value>,
507    #[serde(skip_serializing_if = "Option::is_none")]
508    pub response_format: Option<Value>,
509    #[serde(skip_serializing_if = "Option::is_none")]
510    pub metadata: Option<Value>,
511}
512
513#[derive(Debug, Clone, Serialize, Deserialize)]
514#[non_exhaustive]
515pub struct ErrorBody {
516    pub code: String,
517    pub message: String,
518    #[serde(skip_serializing_if = "Option::is_none")]
519    pub stacktrace: Option<String>,
520}
521
522#[cfg(test)]
523mod tests {
524    use std::collections::HashMap;
525
526    use super::*;
527
528    #[test]
529    fn register_function_to_message_and_serializes_type() {
530        let msg = RegisterFunctionMessage {
531            id: "functions.echo".to_string(),
532            description: Some("Echo function".to_string()),
533            request_format: None,
534            response_format: None,
535            metadata: None,
536            invocation: None,
537        };
538
539        let message = msg.to_message();
540        match &message {
541            Message::RegisterFunction {
542                id, description, ..
543            } => {
544                assert_eq!(id, "functions.echo");
545                assert_eq!(description.as_deref(), Some("Echo function"));
546            }
547            _ => panic!("unexpected message variant"),
548        }
549
550        let serialized = serde_json::to_value(&message).unwrap();
551        assert_eq!(serialized["type"], "registerfunction");
552        assert_eq!(serialized["id"], "functions.echo");
553        assert_eq!(serialized["description"], "Echo function");
554    }
555
556    #[test]
557    fn register_http_function_serializes_invocation() {
558        use iii_helpers::http::{HttpInvocationConfig, HttpMethod};
559
560        let msg = RegisterFunctionMessage {
561            id: "external::my_lambda".to_string(),
562            description: None,
563            request_format: None,
564            response_format: None,
565            metadata: None,
566            invocation: Some(HttpInvocationConfig {
567                url: "https://example.com/invoke".to_string(),
568                method: HttpMethod::Post,
569                timeout_ms: Some(30000),
570                headers: HashMap::new(),
571                auth: None,
572            }),
573        };
574
575        let serialized = serde_json::to_value(msg.to_message()).unwrap();
576        assert_eq!(serialized["type"], "registerfunction");
577        assert_eq!(serialized["id"], "external::my_lambda");
578        assert!(serialized["invocation"].is_object());
579        assert_eq!(
580            serialized["invocation"]["url"],
581            "https://example.com/invoke"
582        );
583        assert_eq!(serialized["invocation"]["method"], "POST");
584    }
585}
iii_sdk/protocol.rs

iii_sdk/
protocol.rs
