lash_remote_protocol/
triggers.rs

1//! Trigger envelopes: occurrence emission, subscriptions, and registrations.
2
3use std::collections::BTreeMap;
4
5use schemars::JsonSchema;
6use serde::{Deserialize, Serialize};
7
8use crate::processes::{
9    RemoteProcessDefinitionIdentity, RemoteProcessEventType, RemoteProcessExecutionEnvRef,
10    RemoteProcessIdentity, RemoteProcessInput, RemoteProcessOriginator, RemoteSessionScope,
11};
12use crate::registry_errors::{RemoteProtocolError, require_non_empty};
13use crate::{REMOTE_PROTOCOL_VERSION, ensure_protocol_version};
14
15#[derive(Clone, Debug, PartialEq, Serialize, Deserialize, JsonSchema)]
16pub struct RemoteTriggerOccurrenceRequest {
17    pub protocol_version: u32,
18    pub source_type: String,
19    pub source_key: String,
20    #[serde(default)]
21    pub payload: serde_json::Value,
22    pub idempotency_key: String,
23    #[serde(default, skip_serializing_if = "Option::is_none")]
24    pub source: Option<serde_json::Value>,
25}
26
27impl RemoteTriggerOccurrenceRequest {
28    pub fn new(
29        source_type: impl Into<String>,
30        source_key: impl Into<String>,
31        payload: serde_json::Value,
32        idempotency_key: impl Into<String>,
33    ) -> Self {
34        Self {
35            protocol_version: REMOTE_PROTOCOL_VERSION,
36            source_type: source_type.into(),
37            source_key: source_key.into(),
38            payload,
39            idempotency_key: idempotency_key.into(),
40            source: None,
41        }
42    }
43
44    pub fn with_source(mut self, source: serde_json::Value) -> Self {
45        self.source = Some(source);
46        self
47    }
48
49    pub fn validate(&self) -> Result<(), RemoteProtocolError> {
50        ensure_protocol_version(self.protocol_version)?;
51        require_non_empty(
52            "RemoteTriggerOccurrenceRequest",
53            "source_type",
54            &self.source_type,
55        )?;
56        require_non_empty(
57            "RemoteTriggerOccurrenceRequest",
58            "source_key",
59            &self.source_key,
60        )?;
61        require_non_empty(
62            "RemoteTriggerOccurrenceRequest",
63            "idempotency_key",
64            &self.idempotency_key,
65        )
66    }
67}
68
69#[derive(Clone, Debug, PartialEq, Serialize, Deserialize, JsonSchema)]
70pub struct RemoteTriggerOccurrenceRecord {
71    pub occurrence_id: String,
72    pub source_type: String,
73    pub source_key: String,
74    #[serde(default)]
75    pub payload: serde_json::Value,
76    pub idempotency_key: String,
77    #[serde(default, skip_serializing_if = "Option::is_none")]
78    pub source: Option<serde_json::Value>,
79    pub occurred_at_ms: u64,
80}
81
82#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize, JsonSchema)]
83#[serde(rename_all = "snake_case")]
84pub enum RemoteTriggerDeliveryEmitOutcome {
85    Started,
86    AlreadyReserved,
87    Failed { reason: String },
88}
89
90#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize, JsonSchema)]
91pub struct RemoteTriggerDeliveryEmitReport {
92    pub occurrence_id: String,
93    pub subscription_id: String,
94    pub process_id: String,
95    pub outcome: RemoteTriggerDeliveryEmitOutcome,
96}
97
98#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize, JsonSchema)]
99pub struct RemoteTriggerEmitReport {
100    pub protocol_version: u32,
101    #[serde(default, skip_serializing_if = "String::is_empty")]
102    pub occurrence_id: String,
103    #[serde(default, skip_serializing_if = "Vec::is_empty")]
104    pub deliveries: Vec<RemoteTriggerDeliveryEmitReport>,
105}
106
107impl RemoteTriggerEmitReport {
108    pub fn validate(&self) -> Result<(), RemoteProtocolError> {
109        ensure_protocol_version(self.protocol_version)
110    }
111}
112
113#[derive(Clone, Debug, PartialEq, Serialize, Deserialize, JsonSchema)]
114pub struct RemoteTriggerSubscriptionFilter {
115    pub protocol_version: u32,
116    #[serde(default, skip_serializing_if = "Option::is_none")]
117    pub registrant_scope_id: Option<String>,
118    #[serde(default, skip_serializing_if = "Option::is_none")]
119    pub session_id: Option<String>,
120    #[serde(default, skip_serializing_if = "Option::is_none")]
121    pub handle: Option<String>,
122    #[serde(default, skip_serializing_if = "Option::is_none")]
123    pub name: Option<String>,
124    #[serde(default, skip_serializing_if = "Option::is_none")]
125    pub source_type: Option<String>,
126    #[serde(default, skip_serializing_if = "Option::is_none")]
127    pub source_key: Option<String>,
128    #[serde(default, skip_serializing_if = "Option::is_none")]
129    pub target: Option<RemoteProcessDefinitionIdentity>,
130    #[serde(default, skip_serializing_if = "Option::is_none")]
131    pub enabled: Option<bool>,
132}
133
134impl Default for RemoteTriggerSubscriptionFilter {
135    fn default() -> Self {
136        Self {
137            protocol_version: REMOTE_PROTOCOL_VERSION,
138            registrant_scope_id: None,
139            session_id: None,
140            handle: None,
141            name: None,
142            source_type: None,
143            source_key: None,
144            target: None,
145            enabled: None,
146        }
147    }
148}
149
150impl RemoteTriggerSubscriptionFilter {
151    pub fn for_session(session_id: impl Into<String>) -> Self {
152        Self {
153            protocol_version: REMOTE_PROTOCOL_VERSION,
154            session_id: Some(session_id.into()),
155            ..Self::default()
156        }
157    }
158
159    pub fn for_registrant_scope(scope_id: impl Into<String>) -> Self {
160        Self {
161            protocol_version: REMOTE_PROTOCOL_VERSION,
162            registrant_scope_id: Some(scope_id.into()),
163            ..Self::default()
164        }
165    }
166
167    pub fn for_source_type(source_type: impl Into<String>) -> Self {
168        Self {
169            protocol_version: REMOTE_PROTOCOL_VERSION,
170            source_type: Some(source_type.into()),
171            ..Self::default()
172        }
173    }
174
175    pub fn validate(&self) -> Result<(), RemoteProtocolError> {
176        ensure_protocol_version(self.protocol_version)
177    }
178}
179
180#[derive(Clone, Debug, PartialEq, Serialize, Deserialize, JsonSchema)]
181pub struct RemoteTriggerRegistration {
182    pub handle: String,
183    pub source_key: String,
184    #[serde(default, skip_serializing_if = "Option::is_none")]
185    pub name: Option<String>,
186    pub source_type: String,
187    #[serde(default)]
188    pub source: serde_json::Value,
189    pub target: RemoteTriggerTargetSummary,
190    #[serde(default = "default_true")]
191    pub enabled: bool,
192}
193
194#[derive(Clone, Debug, PartialEq, Serialize, Deserialize, JsonSchema)]
195pub struct RemoteTriggerTargetSummary {
196    #[serde(default, skip_serializing_if = "Option::is_none")]
197    pub label: Option<String>,
198    pub identity: RemoteProcessIdentity,
199    pub input: RemoteProcessInput,
200    #[serde(default)]
201    pub inputs: RemoteTriggerInputTemplate,
202}
203
204fn default_true() -> bool {
205    true
206}
207
208#[derive(Clone, Debug, Default, PartialEq, Serialize, Deserialize, JsonSchema)]
209#[serde(transparent)]
210pub struct RemoteTriggerInputTemplate {
211    pub entries: BTreeMap<String, RemoteTriggerInputBinding>,
212}
213
214impl RemoteTriggerInputTemplate {
215    pub fn new(entries: BTreeMap<String, RemoteTriggerInputBinding>) -> Self {
216        Self { entries }
217    }
218
219    pub fn validate(&self, type_name: &'static str) -> Result<(), RemoteProtocolError> {
220        for name in self.entries.keys() {
221            require_non_empty(type_name, "input_template key", name)?;
222        }
223        Ok(())
224    }
225}
226
227#[derive(Clone, Debug, PartialEq, Serialize, Deserialize, JsonSchema)]
228#[serde(tag = "kind", rename_all = "snake_case")]
229pub enum RemoteTriggerInputBinding {
230    Event,
231    Fixed { value: serde_json::Value },
232}
233
234#[derive(Clone, Debug, PartialEq, Serialize, Deserialize, JsonSchema)]
235pub struct RemoteTriggerSubscriptionDraft {
236    pub protocol_version: u32,
237    pub registrant: RemoteProcessOriginator,
238    pub env_ref: RemoteProcessExecutionEnvRef,
239    #[serde(default, skip_serializing_if = "Option::is_none")]
240    pub wake_target: Option<RemoteSessionScope>,
241    #[serde(default, skip_serializing_if = "Option::is_none")]
242    pub name: Option<String>,
243    pub source_type: String,
244    pub source_key: String,
245    #[serde(default)]
246    pub source: serde_json::Value,
247    #[serde(default)]
248    pub payload_schema: serde_json::Value,
249    pub target: RemoteProcessInput,
250    pub target_identity: RemoteProcessIdentity,
251    #[serde(default, skip_serializing_if = "Vec::is_empty")]
252    pub event_types: Vec<RemoteProcessEventType>,
253    #[serde(default)]
254    pub input_template: RemoteTriggerInputTemplate,
255    #[serde(default, skip_serializing_if = "Option::is_none")]
256    pub target_label: Option<String>,
257}
258
259impl RemoteTriggerSubscriptionDraft {
260    pub fn for_process(
261        registrant: RemoteProcessOriginator,
262        env_ref: RemoteProcessExecutionEnvRef,
263        source_type: impl Into<String>,
264        source_key: impl Into<String>,
265        target: RemoteProcessInput,
266        target_identity: RemoteProcessIdentity,
267    ) -> Self {
268        let target_label = target_identity.label.clone();
269        Self {
270            protocol_version: REMOTE_PROTOCOL_VERSION,
271            registrant,
272            env_ref,
273            wake_target: None,
274            name: None,
275            source_type: source_type.into(),
276            source_key: source_key.into(),
277            source: serde_json::Value::Object(serde_json::Map::new()),
278            payload_schema: serde_json::Value::Object(serde_json::Map::new()),
279            target,
280            target_identity,
281            event_types: Vec::new(),
282            input_template: RemoteTriggerInputTemplate::default(),
283            target_label,
284        }
285    }
286
287    pub fn with_name(mut self, name: impl Into<String>) -> Self {
288        self.name = Some(name.into());
289        self
290    }
291
292    pub fn with_source(mut self, source: serde_json::Value) -> Self {
293        self.source = source;
294        self
295    }
296
297    pub fn with_payload_schema(mut self, payload_schema: serde_json::Value) -> Self {
298        self.payload_schema = payload_schema;
299        self
300    }
301
302    pub fn with_wake_target(mut self, wake_target: RemoteSessionScope) -> Self {
303        self.wake_target = Some(wake_target);
304        self
305    }
306
307    pub fn with_event_types(
308        mut self,
309        event_types: impl IntoIterator<Item = RemoteProcessEventType>,
310    ) -> Self {
311        self.event_types = event_types.into_iter().collect();
312        self
313    }
314
315    pub fn with_input_template(mut self, input_template: RemoteTriggerInputTemplate) -> Self {
316        self.input_template = input_template;
317        self
318    }
319
320    pub fn with_target_label(mut self, target_label: impl Into<String>) -> Self {
321        self.target_label = Some(target_label.into());
322        self
323    }
324
325    pub fn validate(&self) -> Result<(), RemoteProtocolError> {
326        ensure_protocol_version(self.protocol_version)?;
327        self.registrant.validate("RemoteTriggerSubscriptionDraft")?;
328        self.env_ref.validate("RemoteTriggerSubscriptionDraft")?;
329        if let Some(wake_target) = &self.wake_target {
330            wake_target.validate("RemoteTriggerSubscriptionDraft")?;
331        }
332        require_non_empty(
333            "RemoteTriggerSubscriptionDraft",
334            "source_type",
335            &self.source_type,
336        )?;
337        require_non_empty(
338            "RemoteTriggerSubscriptionDraft",
339            "source_key",
340            &self.source_key,
341        )?;
342        self.target.validate("RemoteTriggerSubscriptionDraft")?;
343        self.target_identity
344            .validate("RemoteTriggerSubscriptionDraft")?;
345        for event_type in &self.event_types {
346            event_type.validate("RemoteTriggerSubscriptionDraft")?;
347        }
348        validate_remote_trigger_target_label(
349            "RemoteTriggerSubscriptionDraft",
350            self.target_label.as_deref(),
351            self.target_identity.label.as_deref(),
352        )?;
353        self.input_template
354            .validate("RemoteTriggerSubscriptionDraft")
355    }
356}
357
358#[derive(Clone, Debug, PartialEq, Serialize, Deserialize, JsonSchema)]
359pub struct RemoteTriggerSubscriptionRecord {
360    pub subscription_id: String,
361    pub registrant: RemoteProcessOriginator,
362    pub env_ref: RemoteProcessExecutionEnvRef,
363    #[serde(default, skip_serializing_if = "Option::is_none")]
364    pub wake_target: Option<RemoteSessionScope>,
365    pub handle: String,
366    #[serde(default, skip_serializing_if = "Option::is_none")]
367    pub name: Option<String>,
368    pub source_type: String,
369    pub source_key: String,
370    #[serde(default)]
371    pub source: serde_json::Value,
372    #[serde(default)]
373    pub payload_schema: serde_json::Value,
374    pub target: RemoteProcessInput,
375    pub target_identity: RemoteProcessIdentity,
376    #[serde(default, skip_serializing_if = "Vec::is_empty")]
377    pub event_types: Vec<RemoteProcessEventType>,
378    #[serde(default)]
379    pub input_template: RemoteTriggerInputTemplate,
380    #[serde(default, skip_serializing_if = "Option::is_none")]
381    pub target_label: Option<String>,
382    #[serde(default = "default_true")]
383    pub enabled: bool,
384    pub created_at_ms: u64,
385    pub updated_at_ms: u64,
386}
387
388impl RemoteTriggerSubscriptionRecord {
389    pub fn validate(&self, type_name: &'static str) -> Result<(), RemoteProtocolError> {
390        require_non_empty(type_name, "subscription_id", &self.subscription_id)?;
391        self.registrant.validate(type_name)?;
392        self.env_ref.validate(type_name)?;
393        if let Some(wake_target) = &self.wake_target {
394            wake_target.validate(type_name)?;
395        }
396        require_non_empty(type_name, "handle", &self.handle)?;
397        require_non_empty(type_name, "source_type", &self.source_type)?;
398        require_non_empty(type_name, "source_key", &self.source_key)?;
399        self.target.validate(type_name)?;
400        self.target_identity.validate(type_name)?;
401        for event_type in &self.event_types {
402            event_type.validate(type_name)?;
403        }
404        validate_remote_trigger_target_label(
405            type_name,
406            self.target_label.as_deref(),
407            self.target_identity.label.as_deref(),
408        )?;
409        self.input_template.validate(type_name)
410    }
411}
412
413fn validate_remote_trigger_target_label(
414    type_name: &'static str,
415    target_label: Option<&str>,
416    identity_label: Option<&str>,
417) -> Result<(), RemoteProtocolError> {
418    match (target_label, identity_label) {
419        (Some(target_label), Some(identity_label)) if target_label != identity_label => {
420            Err(RemoteProtocolError::InvalidEnvelope {
421                type_name,
422                message: "target_label must match target_identity.label when both are present"
423                    .to_string(),
424            })
425        }
426        _ => Ok(()),
427    }
428}
429
430#[derive(Clone, Debug, PartialEq, Serialize, Deserialize, JsonSchema)]
431pub struct RemoteTriggerRegisterSubscriptionRequest {
432    pub protocol_version: u32,
433    pub draft: RemoteTriggerSubscriptionDraft,
434}
435
436impl RemoteTriggerRegisterSubscriptionRequest {
437    pub fn validate(&self) -> Result<(), RemoteProtocolError> {
438        ensure_protocol_version(self.protocol_version)?;
439        if self.draft.protocol_version != self.protocol_version {
440            return Err(RemoteProtocolError::MismatchedNestedProtocolVersion {
441                parent: "RemoteTriggerRegisterSubscriptionRequest",
442                child: "draft",
443                parent_version: self.protocol_version,
444                child_version: self.draft.protocol_version,
445            });
446        }
447        self.draft.validate()
448    }
449}
450
451#[derive(Clone, Debug, PartialEq, Serialize, Deserialize, JsonSchema)]
452pub struct RemoteTriggerRegisterSubscriptionResult {
453    pub protocol_version: u32,
454    pub record: RemoteTriggerSubscriptionRecord,
455}
456
457impl RemoteTriggerRegisterSubscriptionResult {
458    pub fn validate(&self) -> Result<(), RemoteProtocolError> {
459        ensure_protocol_version(self.protocol_version)?;
460        self.record
461            .validate("RemoteTriggerRegisterSubscriptionResult")
462    }
463}
464
465#[derive(Clone, Debug, PartialEq, Serialize, Deserialize, JsonSchema)]
466pub struct RemoteTriggerListSubscriptionsResponse {
467    pub protocol_version: u32,
468    #[serde(default)]
469    pub subscriptions: Vec<RemoteTriggerSubscriptionRecord>,
470}
471
472impl RemoteTriggerListSubscriptionsResponse {
473    pub fn validate(&self) -> Result<(), RemoteProtocolError> {
474        ensure_protocol_version(self.protocol_version)?;
475        for record in &self.subscriptions {
476            record.validate("RemoteTriggerListSubscriptionsResponse")?;
477        }
478        Ok(())
479    }
480}
481
482#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize, JsonSchema)]
483pub struct RemoteTriggerCancelSubscriptionRequest {
484    pub protocol_version: u32,
485    pub session_id: String,
486    pub handle: String,
487}
488
489impl RemoteTriggerCancelSubscriptionRequest {
490    pub fn validate(&self) -> Result<(), RemoteProtocolError> {
491        ensure_protocol_version(self.protocol_version)?;
492        require_non_empty(
493            "RemoteTriggerCancelSubscriptionRequest",
494            "session_id",
495            &self.session_id,
496        )?;
497        require_non_empty(
498            "RemoteTriggerCancelSubscriptionRequest",
499            "handle",
500            &self.handle,
501        )
502    }
503}
504
505#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize, JsonSchema)]
506pub struct RemoteTriggerCancelSubscriptionResult {
507    pub protocol_version: u32,
508    pub session_id: String,
509    pub handle: String,
510    pub cancelled: bool,
511}
512
513impl RemoteTriggerCancelSubscriptionResult {
514    pub fn validate(&self) -> Result<(), RemoteProtocolError> {
515        ensure_protocol_version(self.protocol_version)?;
516        require_non_empty(
517            "RemoteTriggerCancelSubscriptionResult",
518            "session_id",
519            &self.session_id,
520        )?;
521        require_non_empty(
522            "RemoteTriggerCancelSubscriptionResult",
523            "handle",
524            &self.handle,
525        )
526    }
527}
lash_remote_protocol/triggers.rs

lash_remote_protocol/
triggers.rs
