lash_core/
triggers.rs

1use std::collections::BTreeMap;
2use std::sync::{Arc, Mutex};
3
4use serde::{Deserialize, Serialize};
5
6use crate::plugin::PluginError;
7
8mod memory;
9mod mutation;
10mod router;
11
12pub use memory::InMemoryTriggerStore;
13use memory::{
14    InMemoryTriggerDeliveryRecord, InMemoryTriggerEventState, apply_in_memory_trigger_command,
15};
16pub use mutation::evaluate_trigger_mutation;
17use mutation::{
18    ensure_live_revision, mutate_enabled, subscription_conflict, subscription_record_from_draft,
19};
20pub use router::*;
21use router::{default_enabled, reserve_in_memory_for_occurrence};
22
23#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
24pub struct TriggerEvent {
25    pub resource_type: String,
26    pub alias: String,
27    pub event: String,
28    pub payload_schema: crate::LashSchema,
29}
30
31impl TriggerEvent {
32    pub fn new(
33        resource_type: impl Into<String>,
34        alias: impl Into<String>,
35        event: impl Into<String>,
36        payload_schema: crate::LashSchema,
37    ) -> Self {
38        Self {
39            resource_type: resource_type.into(),
40            alias: alias.into(),
41            event: event.into(),
42            payload_schema,
43        }
44    }
45
46    pub fn payload_schema(&self) -> &crate::LashSchema {
47        &self.payload_schema
48    }
49
50    pub fn key(&self) -> TriggerEventKey {
51        TriggerEventKey {
52            resource_type: self.resource_type.clone(),
53            alias: self.alias.clone(),
54            event: self.event.clone(),
55        }
56    }
57
58    pub fn source_type(&self) -> String {
59        trigger_event_type(&self.alias, &self.event)
60    }
61}
62
63#[derive(Clone, Debug, PartialEq, Eq, PartialOrd, Ord, Serialize, Deserialize)]
64pub struct TriggerEventKey {
65    pub resource_type: String,
66    pub alias: String,
67    pub event: String,
68}
69
70impl TriggerEventKey {
71    pub fn new(
72        resource_type: impl Into<String>,
73        alias: impl Into<String>,
74        event: impl Into<String>,
75    ) -> Self {
76        Self {
77            resource_type: resource_type.into(),
78            alias: alias.into(),
79            event: event.into(),
80        }
81    }
82
83    pub fn source_type(&self) -> String {
84        trigger_event_type(&self.alias, &self.event)
85    }
86}
87
88pub fn trigger_event_type(alias: &str, event: &str) -> String {
89    format!("{alias}.{event}")
90}
91
92#[derive(Clone, Debug, Default, PartialEq, Serialize, Deserialize)]
93pub struct TriggerEventCatalog {
94    events: BTreeMap<TriggerEventKey, TriggerEvent>,
95}
96
97impl TriggerEventCatalog {
98    pub fn new() -> Self {
99        Self::default()
100    }
101
102    pub fn declare(&mut self, event: TriggerEvent) -> Result<(), String> {
103        let key = event.key();
104        if self.events.contains_key(&key) {
105            return Err(format!(
106                "duplicate trigger occurrence `{}.{}.{}`",
107                key.resource_type, key.alias, key.event
108            ));
109        }
110        let source_type = event.source_type();
111        if let Some(existing) = self
112            .events
113            .values()
114            .find(|existing| existing.source_type() == source_type)
115        {
116            return Err(format!(
117                "duplicate trigger source `{source_type}` declared by `{}.{}.{}` and `{}.{}.{}`",
118                existing.resource_type,
119                existing.alias,
120                existing.event,
121                key.resource_type,
122                key.alias,
123                key.event
124            ));
125        }
126        self.events.insert(key, event);
127        Ok(())
128    }
129
130    pub fn from_events(events: impl IntoIterator<Item = TriggerEvent>) -> Result<Self, String> {
131        let mut catalog = Self::new();
132        for event in events {
133            catalog.declare(event)?;
134        }
135        Ok(catalog)
136    }
137
138    pub fn get(&self, resource_type: &str, alias: &str, event: &str) -> Option<&TriggerEvent> {
139        self.events
140            .get(&TriggerEventKey::new(resource_type, alias, event))
141    }
142
143    pub fn is_empty(&self) -> bool {
144        self.events.is_empty()
145    }
146
147    pub fn events(&self) -> impl Iterator<Item = &TriggerEvent> {
148        self.events.values()
149    }
150}
151
152#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
153#[serde(rename_all = "snake_case")]
154pub enum TriggerDeliveryEmitOutcome {
155    Started,
156    AlreadyReserved,
157    Failed { reason: String },
158}
159
160#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
161pub struct TriggerDeliveryEmitReport {
162    pub occurrence_id: String,
163    pub subscription_id: String,
164    pub process_id: String,
165    pub outcome: TriggerDeliveryEmitOutcome,
166}
167
168#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)]
169pub struct TriggerEmitReport {
170    #[serde(default, skip_serializing_if = "String::is_empty")]
171    pub occurrence_id: String,
172    #[serde(default, skip_serializing_if = "Vec::is_empty")]
173    pub deliveries: Vec<TriggerDeliveryEmitReport>,
174}
175
176impl TriggerEmitReport {
177    pub fn empty() -> Self {
178        Self::default()
179    }
180
181    fn new(occurrence_id: String, deliveries: Vec<TriggerDeliveryEmitReport>) -> Self {
182        Self {
183            occurrence_id,
184            deliveries,
185        }
186    }
187
188    pub fn started_process_ids(&self) -> Vec<String> {
189        self.deliveries
190            .iter()
191            .filter(|delivery| delivery.outcome == TriggerDeliveryEmitOutcome::Started)
192            .map(|delivery| delivery.process_id.clone())
193            .collect()
194    }
195}
196
197#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
198pub struct TriggerOccurrenceRequest {
199    pub source_type: String,
200    pub source_key: String,
201    #[serde(default)]
202    pub payload: serde_json::Value,
203    pub idempotency_key: String,
204    #[serde(default, skip_serializing_if = "Option::is_none")]
205    pub source: Option<serde_json::Value>,
206    /// Optional host routing scope. When present, only subscriptions
207    /// registered by this session can reserve deliveries for the occurrence.
208    #[serde(default, skip_serializing_if = "Option::is_none")]
209    pub session_id: Option<String>,
210}
211
212impl TriggerOccurrenceRequest {
213    pub fn new(
214        source_type: impl Into<String>,
215        source_key: impl Into<String>,
216        payload: serde_json::Value,
217        idempotency_key: impl Into<String>,
218    ) -> Self {
219        Self {
220            source_type: source_type.into(),
221            source_key: source_key.into(),
222            payload,
223            idempotency_key: idempotency_key.into(),
224            source: None,
225            session_id: None,
226        }
227    }
228
229    pub fn with_source(mut self, source: serde_json::Value) -> Self {
230        self.source = Some(source);
231        self
232    }
233
234    pub fn for_session(mut self, session_id: impl Into<String>) -> Self {
235        self.session_id = Some(session_id.into());
236        self
237    }
238}
239
240#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
241pub struct TriggerOccurrenceRecord {
242    pub occurrence_id: String,
243    pub source_type: String,
244    pub source_key: String,
245    #[serde(default)]
246    pub payload: serde_json::Value,
247    pub idempotency_key: String,
248    #[serde(default, skip_serializing_if = "Option::is_none")]
249    pub source: Option<serde_json::Value>,
250    #[serde(default, skip_serializing_if = "Option::is_none")]
251    pub session_id: Option<String>,
252    pub occurred_at_ms: u64,
253}
254
255#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)]
256pub struct TriggerOccurrenceFilter {
257    #[serde(default, skip_serializing_if = "Option::is_none")]
258    pub source_type: Option<String>,
259    #[serde(default, skip_serializing_if = "Option::is_none")]
260    pub source_key: Option<String>,
261    #[serde(default, skip_serializing_if = "Option::is_none")]
262    pub occurred_at_start_ms: Option<u64>,
263    #[serde(default, skip_serializing_if = "Option::is_none")]
264    pub occurred_at_end_ms: Option<u64>,
265}
266
267impl TriggerOccurrenceFilter {
268    pub fn for_source(source_type: impl Into<String>, source_key: impl Into<String>) -> Self {
269        Self {
270            source_type: Some(source_type.into()),
271            source_key: Some(source_key.into()),
272            ..Self::default()
273        }
274    }
275
276    pub fn matches(&self, record: &TriggerOccurrenceRecord) -> bool {
277        self.source_type
278            .as_deref()
279            .is_none_or(|source_type| record.source_type == source_type)
280            && self
281                .source_key
282                .as_deref()
283                .is_none_or(|source_key| record.source_key == source_key)
284            && self
285                .occurred_at_start_ms
286                .is_none_or(|start_ms| record.occurred_at_ms >= start_ms)
287            && self
288                .occurred_at_end_ms
289                .is_none_or(|end_ms| record.occurred_at_ms < end_ms)
290    }
291}
292
293#[derive(Clone, Debug, PartialEq, Eq, PartialOrd, Ord, Hash, Serialize, Deserialize)]
294#[serde(transparent)]
295pub struct TriggerEventType(String);
296
297impl TriggerEventType {
298    pub fn new(value: impl Into<String>) -> Self {
299        Self(value.into())
300    }
301
302    pub fn as_str(&self) -> &str {
303        &self.0
304    }
305}
306
307impl From<String> for TriggerEventType {
308    fn from(value: String) -> Self {
309        Self::new(value)
310    }
311}
312
313impl From<&str> for TriggerEventType {
314    fn from(value: &str) -> Self {
315        Self::new(value)
316    }
317}
318
319impl AsRef<str> for TriggerEventType {
320    fn as_ref(&self) -> &str {
321        self.as_str()
322    }
323}
324
325impl std::fmt::Display for TriggerEventType {
326    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
327        formatter.write_str(self.as_str())
328    }
329}
330
331#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
332pub struct TriggerRegistration {
333    pub subscription_key: String,
334    pub incarnation: String,
335    pub revision: u64,
336    pub registrant: crate::ProcessOriginator,
337    pub manifest_membership: TriggerManifestMembership,
338    pub source_key: String,
339    #[serde(default, skip_serializing_if = "Option::is_none")]
340    pub name: Option<String>,
341    pub source_type: TriggerEventType,
342    pub source: serde_json::Value,
343    pub target: TriggerTargetSummary,
344    #[serde(default = "default_enabled")]
345    pub enabled: bool,
346}
347
348#[derive(Clone, Copy, Debug, Default, PartialEq, Eq, Serialize, Deserialize)]
349#[serde(rename_all = "snake_case")]
350pub enum TriggerManifestMembership {
351    PresentInCurrentArtifact,
352    Orphaned,
353    #[default]
354    Unknown,
355}
356
357#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
358pub struct TriggerTargetSummary {
359    pub label: Option<String>,
360    pub identity: crate::ProcessIdentity,
361    pub input: crate::ProcessInput,
362    #[serde(default, skip_serializing_if = "std::collections::BTreeMap::is_empty")]
363    pub inputs: BTreeMap<String, TriggerInputBinding>,
364}
365
366#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
367#[serde(tag = "type", rename_all = "snake_case")]
368pub enum TriggerInputBinding {
369    Event,
370    Fixed { value: serde_json::Value },
371}
372
373#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
374pub struct TriggerSubscriptionDraft {
375    pub subscription_key: String,
376    pub env_ref: crate::ProcessExecutionEnvRef,
377    #[serde(default, skip_serializing_if = "Option::is_none")]
378    pub wake_target: Option<crate::SessionScope>,
379    #[serde(default, skip_serializing_if = "Option::is_none")]
380    pub name: Option<String>,
381    pub source_type: String,
382    pub source_key: String,
383    pub source: serde_json::Value,
384    pub payload_schema: crate::LashSchema,
385    pub target: crate::ProcessInput,
386    pub target_identity: crate::ProcessIdentity,
387    #[serde(default)]
388    pub event_types: Vec<crate::ProcessEventType>,
389    #[serde(default)]
390    pub input_template: BTreeMap<String, TriggerInputBinding>,
391    #[serde(default, skip_serializing_if = "Option::is_none")]
392    pub target_label: Option<String>,
393}
394
395impl TriggerSubscriptionDraft {
396    pub fn for_process(
397        subscription_key: impl Into<String>,
398        env_ref: crate::ProcessExecutionEnvRef,
399        source_type: impl Into<String>,
400        source_key: impl Into<String>,
401        target: crate::ProcessInput,
402        target_identity: crate::ProcessIdentity,
403    ) -> Self {
404        let target_label = target_identity.label.clone();
405        Self {
406            subscription_key: subscription_key.into(),
407            env_ref,
408            wake_target: None,
409            name: None,
410            source_type: source_type.into(),
411            source_key: source_key.into(),
412            source: serde_json::Value::Object(serde_json::Map::new()),
413            payload_schema: crate::LashSchema::new(serde_json::Value::Object(
414                serde_json::Map::new(),
415            )),
416            target,
417            target_identity,
418            event_types: Vec::new(),
419            input_template: BTreeMap::new(),
420            target_label,
421        }
422    }
423
424    pub fn with_name(mut self, name: impl Into<String>) -> Self {
425        self.name = Some(name.into());
426        self
427    }
428
429    pub fn with_source(mut self, source: serde_json::Value) -> Self {
430        self.source = source;
431        self
432    }
433
434    pub fn with_payload_schema(mut self, payload_schema: crate::LashSchema) -> Self {
435        self.payload_schema = payload_schema;
436        self
437    }
438
439    pub fn with_wake_target(mut self, wake_target: crate::SessionScope) -> Self {
440        self.wake_target = Some(wake_target);
441        self
442    }
443
444    pub fn with_event_types(
445        mut self,
446        event_types: impl IntoIterator<Item = crate::ProcessEventType>,
447    ) -> Self {
448        self.event_types = event_types.into_iter().collect();
449        self
450    }
451
452    pub fn with_input_template(
453        mut self,
454        input_template: BTreeMap<String, TriggerInputBinding>,
455    ) -> Self {
456        self.input_template = input_template;
457        self
458    }
459
460    pub fn with_target_label(mut self, target_label: impl Into<String>) -> Self {
461        self.target_label = Some(target_label.into());
462        self
463    }
464
465    pub fn validate(&self) -> Result<(), PluginError> {
466        validate_subscription_key(&self.subscription_key, false)?;
467        validate_trigger_subscription_target_label(
468            self.target_label.as_deref(),
469            self.target_identity.label.as_deref(),
470        )
471    }
472}
473
474pub const INTERNAL_TRIGGER_KEY_PREFIX: &str = "lash.internal/";
475
476pub fn validate_subscription_key(key: &str, internal: bool) -> Result<(), PluginError> {
477    if key.trim().is_empty() {
478        return Err(PluginError::Session(
479            "trigger subscription requires subscription_key".to_string(),
480        ));
481    }
482    if !internal && key.starts_with(INTERNAL_TRIGGER_KEY_PREFIX) {
483        return Err(PluginError::Session(format!(
484            "trigger subscription key `{key}` uses reserved prefix `{INTERNAL_TRIGGER_KEY_PREFIX}`"
485        )));
486    }
487    Ok(())
488}
489
490#[derive(Clone, Debug, PartialEq, Eq, PartialOrd, Ord, Hash, Serialize, Deserialize)]
491#[serde(tag = "type", rename_all = "snake_case")]
492pub enum TriggerOwnerScope {
493    Session { session_id: String },
494    Host { binding_id: String },
495    Platform,
496}
497
498impl TriggerOwnerScope {
499    pub fn session(session_id: impl Into<String>) -> Self {
500        Self::Session {
501            session_id: session_id.into(),
502        }
503    }
504
505    pub fn host(binding_id: impl Into<String>) -> Result<Self, PluginError> {
506        let binding_id = binding_id.into();
507        if binding_id.trim().is_empty() {
508            return Err(PluginError::Session(
509                "trigger host owner requires a non-empty binding id".to_string(),
510            ));
511        }
512        Ok(Self::Host { binding_id })
513    }
514
515    pub fn namespace(&self) -> String {
516        match self {
517            Self::Session { session_id } => format!("session:{session_id}"),
518            Self::Host { binding_id } => format!("host:{binding_id}"),
519            Self::Platform => "host".to_string(),
520        }
521    }
522
523    pub fn session_id(&self) -> Option<&str> {
524        match self {
525            Self::Session { session_id } => Some(session_id),
526            Self::Host { .. } | Self::Platform => None,
527        }
528    }
529}
530
531#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
532pub struct TriggerSubscriptionRecord {
533    pub subscription_id: String,
534    pub owner_scope: TriggerOwnerScope,
535    pub subscription_key: String,
536    pub incarnation: String,
537    pub revision: u64,
538    pub definition_hash: String,
539    pub registrant: crate::ProcessOriginator,
540    pub env_ref: crate::ProcessExecutionEnvRef,
541    #[serde(default, skip_serializing_if = "Option::is_none")]
542    pub wake_target: Option<crate::SessionScope>,
543    #[serde(default, skip_serializing_if = "Option::is_none")]
544    pub name: Option<String>,
545    pub source_type: String,
546    pub source_key: String,
547    pub source: serde_json::Value,
548    pub payload_schema: crate::LashSchema,
549    pub target: crate::ProcessInput,
550    pub target_identity: crate::ProcessIdentity,
551    #[serde(default)]
552    pub event_types: Vec<crate::ProcessEventType>,
553    #[serde(default)]
554    pub input_template: BTreeMap<String, TriggerInputBinding>,
555    #[serde(default, skip_serializing_if = "Option::is_none")]
556    pub target_label: Option<String>,
557    #[serde(default = "default_enabled")]
558    pub enabled: bool,
559    #[serde(default)]
560    pub tombstoned: bool,
561    #[serde(default, skip_serializing_if = "Option::is_none")]
562    pub deleted_at_ms: Option<u64>,
563    pub created_at_ms: u64,
564    pub updated_at_ms: u64,
565}
566
567impl TriggerSubscriptionRecord {
568    pub fn registrant_scope_id(&self) -> String {
569        self.owner_scope.namespace()
570    }
571
572    pub fn registrant_session_id(&self) -> Option<&str> {
573        self.owner_scope.session_id()
574    }
575}
576
577fn validate_trigger_subscription_target_label(
578    target_label: Option<&str>,
579    identity_label: Option<&str>,
580) -> Result<(), PluginError> {
581    match (target_label, identity_label) {
582        (Some(target_label), Some(identity_label)) if target_label != identity_label => {
583            Err(PluginError::Session(
584                "trigger target_label must match target_identity.label when both are present"
585                    .to_string(),
586            ))
587        }
588        _ => Ok(()),
589    }
590}
591
592impl From<&TriggerSubscriptionRecord> for TriggerRegistration {
593    fn from(route: &TriggerSubscriptionRecord) -> Self {
594        Self::from_record_with_manifest(route, None)
595    }
596}
597
598impl TriggerRegistration {
599    pub fn from_record_with_manifest(
600        route: &TriggerSubscriptionRecord,
601        manifest_keys: Option<&std::collections::BTreeSet<String>>,
602    ) -> Self {
603        Self {
604            subscription_key: route.subscription_key.clone(),
605            incarnation: route.incarnation.clone(),
606            revision: route.revision,
607            registrant: route.registrant.clone(),
608            manifest_membership: manifest_keys.map_or(TriggerManifestMembership::Unknown, |keys| {
609                if keys.contains(&route.subscription_key) {
610                    TriggerManifestMembership::PresentInCurrentArtifact
611                } else {
612                    TriggerManifestMembership::Orphaned
613                }
614            }),
615            source_key: route.source_key.clone(),
616            name: route.name.clone(),
617            source_type: TriggerEventType::new(route.source_type.clone()),
618            source: route.source.clone(),
619            target: TriggerTargetSummary {
620                label: route.target_label.clone(),
621                identity: route.target_identity.clone(),
622                input: route.target.clone(),
623                inputs: route.input_template.clone(),
624            },
625            enabled: route.enabled,
626        }
627    }
628}
629
630#[derive(Clone, Debug, Default, PartialEq, Serialize, Deserialize)]
631pub struct TriggerSubscriptionFilter {
632    #[serde(default, skip_serializing_if = "Option::is_none")]
633    pub registrant_scope_id: Option<String>,
634    #[serde(default, skip_serializing_if = "Option::is_none")]
635    pub session_id: Option<String>,
636    #[serde(default, skip_serializing_if = "Option::is_none")]
637    pub subscription_key: Option<String>,
638    #[serde(default, skip_serializing_if = "Option::is_none")]
639    pub name: Option<String>,
640    #[serde(default, skip_serializing_if = "Option::is_none")]
641    pub source_type: Option<String>,
642    #[serde(default, skip_serializing_if = "Option::is_none")]
643    pub source_key: Option<String>,
644    #[serde(default, skip_serializing_if = "Option::is_none")]
645    pub target: Option<serde_json::Value>,
646    #[serde(default, skip_serializing_if = "Option::is_none")]
647    pub enabled: Option<bool>,
648}
649
650impl TriggerSubscriptionFilter {
651    pub fn for_session(session_id: impl Into<String>) -> Self {
652        Self {
653            session_id: Some(session_id.into()),
654            ..Self::default()
655        }
656    }
657
658    pub fn for_registrant_scope(scope_id: impl Into<String>) -> Self {
659        Self {
660            registrant_scope_id: Some(scope_id.into()),
661            ..Self::default()
662        }
663    }
664
665    pub fn for_source_type(source_type: impl Into<String>) -> Self {
666        Self {
667            source_type: Some(source_type.into()),
668            ..Self::default()
669        }
670    }
671
672    pub fn effective_registrant_scope_id(&self) -> Option<String> {
673        self.registrant_scope_id.clone()
674    }
675
676    pub fn matches(&self, record: &TriggerSubscriptionRecord) -> bool {
677        self.effective_registrant_scope_id()
678            .is_none_or(|scope_id| record.registrant_scope_id() == scope_id)
679            && self
680                .session_id
681                .as_deref()
682                .is_none_or(|session_id| record.registrant_session_id() == Some(session_id))
683            && self
684                .subscription_key
685                .as_deref()
686                .is_none_or(|key| record.subscription_key == key)
687            && self
688                .name
689                .as_deref()
690                .is_none_or(|name| record.name.as_deref() == Some(name))
691            && self
692                .source_type
693                .as_deref()
694                .is_none_or(|source_type| record.source_type == source_type)
695            && self
696                .source_key
697                .as_deref()
698                .is_none_or(|source_key| record.source_key == source_key)
699            && self.enabled.is_none_or(|enabled| record.enabled == enabled)
700            && !record.tombstoned
701            && self
702                .target
703                .as_ref()
704                .is_none_or(|target| record.target_identity.definition.as_ref() == Some(target))
705    }
706}
707
708#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)]
709#[serde(rename_all = "snake_case")]
710pub enum TriggerMutationDisposition {
711    Created,
712    Unchanged,
713    Updated,
714    Enabled,
715    Disabled,
716    Deleted,
717    Revived,
718}
719
720#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
721pub struct TriggerMutationReceipt {
722    pub owner_scope: TriggerOwnerScope,
723    pub subscription_key: String,
724    pub subscription_id: String,
725    pub incarnation: String,
726    pub revision: u64,
727    pub definition_hash: String,
728    pub enabled: bool,
729    pub disposition: TriggerMutationDisposition,
730    pub record_snapshot: TriggerSubscriptionRecord,
731}
732
733impl TriggerMutationReceipt {
734    fn from_record(
735        record: TriggerSubscriptionRecord,
736        disposition: TriggerMutationDisposition,
737    ) -> Self {
738        Self {
739            owner_scope: record.owner_scope.clone(),
740            subscription_key: record.subscription_key.clone(),
741            subscription_id: record.subscription_id.clone(),
742            incarnation: record.incarnation.clone(),
743            revision: record.revision,
744            definition_hash: record.definition_hash.clone(),
745            enabled: record.enabled,
746            disposition,
747            record_snapshot: record,
748        }
749    }
750}
751
752#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
753#[serde(tag = "op", rename_all = "snake_case")]
754pub enum TriggerCommand {
755    Register {
756        owner_scope: TriggerOwnerScope,
757        actor: crate::ProcessOriginator,
758        draft: TriggerSubscriptionDraft,
759    },
760    List {
761        owner_scope: TriggerOwnerScope,
762        filter: TriggerSubscriptionFilter,
763    },
764    Update {
765        owner_scope: TriggerOwnerScope,
766        actor: crate::ProcessOriginator,
767        subscription_key: String,
768        draft: TriggerSubscriptionDraft,
769        expected_revision: u64,
770    },
771    Enable {
772        owner_scope: TriggerOwnerScope,
773        actor: crate::ProcessOriginator,
774        subscription_key: String,
775        expected_revision: u64,
776    },
777    Disable {
778        owner_scope: TriggerOwnerScope,
779        actor: crate::ProcessOriginator,
780        subscription_key: String,
781        expected_revision: u64,
782    },
783    Delete {
784        owner_scope: TriggerOwnerScope,
785        actor: crate::ProcessOriginator,
786        subscription_key: String,
787        expected_revision: u64,
788    },
789    Revive {
790        owner_scope: TriggerOwnerScope,
791        actor: crate::ProcessOriginator,
792        subscription_key: String,
793        draft: TriggerSubscriptionDraft,
794        expected_revision: u64,
795    },
796    Prune {
797        owner_scope: TriggerOwnerScope,
798        actor: crate::ProcessOriginator,
799        subscription_keys: Vec<String>,
800    },
801}
802
803impl TriggerCommand {
804    pub fn owner_scope(&self) -> &TriggerOwnerScope {
805        match self {
806            Self::Register { owner_scope, .. }
807            | Self::List { owner_scope, .. }
808            | Self::Update { owner_scope, .. }
809            | Self::Enable { owner_scope, .. }
810            | Self::Disable { owner_scope, .. }
811            | Self::Delete { owner_scope, .. }
812            | Self::Revive { owner_scope, .. }
813            | Self::Prune { owner_scope, .. } => owner_scope,
814        }
815    }
816
817    pub fn subscription_key(&self) -> Option<&str> {
818        match self {
819            Self::Register { draft, .. } => Some(&draft.subscription_key),
820            Self::List { .. } => None,
821            Self::Update {
822                subscription_key, ..
823            }
824            | Self::Enable {
825                subscription_key, ..
826            }
827            | Self::Disable {
828                subscription_key, ..
829            }
830            | Self::Delete {
831                subscription_key, ..
832            }
833            | Self::Revive {
834                subscription_key, ..
835            } => Some(subscription_key),
836            Self::Prune { .. } => None,
837        }
838    }
839
840    pub fn is_mutation(&self) -> bool {
841        !matches!(self, Self::List { .. })
842    }
843}
844
845#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
846#[serde(tag = "type", rename_all = "snake_case")]
847pub enum TriggerCommandOutcome {
848    Mutation {
849        receipt: Box<TriggerMutationReceipt>,
850    },
851    List {
852        records: Vec<TriggerSubscriptionRecord>,
853    },
854    Prune {
855        receipts: Vec<TriggerMutationReceipt>,
856    },
857}
858
859#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize, thiserror::Error)]
860#[serde(tag = "type", rename_all = "snake_case")]
861pub enum TriggerOperationError {
862    #[error(
863        "trigger subscription conflict for `{subscription_key}`: {reason}; existing revision {existing_revision:?}, existing definition {existing_definition_hash:?}, requested definition {requested_definition_hash:?}"
864    )]
865    Conflict {
866        subscription_key: String,
867        existing_revision: Option<u64>,
868        existing_definition_hash: Option<String>,
869        requested_definition_hash: Option<String>,
870        reason: String,
871    },
872    #[error("trigger subscription request is invalid: {message}")]
873    Invalid { message: String },
874    #[error("trigger subscription operation failed: {message}")]
875    Store { message: String },
876}
877
878impl From<PluginError> for TriggerOperationError {
879    fn from(value: PluginError) -> Self {
880        Self::Store {
881            message: value.to_string(),
882        }
883    }
884}
885
886pub type TriggerEffectResult = Result<TriggerCommandOutcome, TriggerOperationError>;
887
888// Measured 112 B on rustc 1.97.0, x86_64-unknown-linux-gnu (FIG-595).
889const _: () = assert!(std::mem::size_of::<TriggerEffectResult>() <= 144);
890
891pub fn evaluate_trigger_prune(
892    records: impl IntoIterator<Item = TriggerSubscriptionRecord>,
893    owner_scope: TriggerOwnerScope,
894    actor: crate::ProcessOriginator,
895    subscription_keys: Vec<String>,
896    now: u64,
897) -> TriggerEffectResult {
898    let requested = subscription_keys
899        .into_iter()
900        .collect::<std::collections::BTreeSet<_>>();
901    let mut receipts = Vec::new();
902    for record in records {
903        if record.owner_scope != owner_scope
904            || record.tombstoned
905            || !requested.contains(&record.subscription_key)
906        {
907            continue;
908        }
909        let command = TriggerCommand::Delete {
910            owner_scope: owner_scope.clone(),
911            actor: actor.clone(),
912            subscription_key: record.subscription_key.clone(),
913            expected_revision: record.revision,
914        };
915        match evaluate_trigger_mutation(Some(record), command, now)?? {
916            TriggerCommandOutcome::Mutation { receipt } => receipts.push(*receipt),
917            TriggerCommandOutcome::List { .. } | TriggerCommandOutcome::Prune { .. } => {
918                unreachable!("delete mutation always returns one receipt")
919            }
920        }
921    }
922    receipts.sort_by(|left, right| left.subscription_key.cmp(&right.subscription_key));
923    Ok(TriggerCommandOutcome::Prune { receipts })
924}
925
926pub fn trigger_command_hash(command: &TriggerCommand) -> Result<String, PluginError> {
927    crate::stable_hash::stable_json_sha256_hex(command)
928        .map_err(|err| PluginError::Session(format!("failed to hash trigger command: {err}")))
929}
930
931pub fn trigger_operation_receipt_id(
932    owner_scope: &TriggerOwnerScope,
933    operation_id: &str,
934) -> Result<String, PluginError> {
935    let digest = crate::stable_hash::stable_json_sha256_hex(&(
936        "lash.trigger-operation-receipt",
937        1_u8,
938        owner_scope,
939        operation_id,
940    ))
941    .map_err(|err| PluginError::Session(format!("failed to hash trigger operation id: {err}")))?;
942    Ok(format!("trigger-operation:v1:sha256:{digest}"))
943}
944
945#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
946pub struct TriggerIngressResult {
947    pub occurrence: TriggerOccurrenceRecord,
948    pub reservations: Vec<TriggerDeliveryReservation>,
949}
950
951#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
952#[serde(rename_all = "snake_case")]
953pub enum TriggerDeliveryReservationStatus {
954    Reserved,
955    AlreadyReserved,
956}
957
958#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
959pub struct TriggerDeliveryReservation {
960    pub occurrence: TriggerOccurrenceRecord,
961    pub subscription: TriggerSubscriptionRecord,
962    pub process_id: String,
963    pub created_at_ms: u64,
964    pub reservation_status: TriggerDeliveryReservationStatus,
965}
966
967impl TriggerDeliveryReservation {
968    fn emit_report(&self, outcome: TriggerDeliveryEmitOutcome) -> TriggerDeliveryEmitReport {
969        TriggerDeliveryEmitReport {
970            occurrence_id: self.occurrence.occurrence_id.clone(),
971            subscription_id: self.subscription.subscription_id.clone(),
972            process_id: self.process_id.clone(),
973            outcome,
974        }
975    }
976}
977
978#[async_trait::async_trait]
979pub trait TriggerStore: Send + Sync {
980    fn durability_tier(&self) -> crate::DurabilityTier {
981        crate::DurabilityTier::Inline
982    }
983
984    async fn execute_command(
985        &self,
986        operation_id: &str,
987        command: TriggerCommand,
988    ) -> Result<TriggerEffectResult, PluginError>;
989
990    async fn list_subscriptions(
991        &self,
992        filter: TriggerSubscriptionFilter,
993    ) -> Result<Vec<TriggerSubscriptionRecord>, PluginError>;
994
995    async fn delete_session_subscriptions(&self, session_id: &str) -> Result<usize, PluginError>;
996
997    async fn ingest_occurrence(
998        &self,
999        request: TriggerOccurrenceRequest,
1000    ) -> Result<TriggerIngressResult, PluginError>;
1001
1002    async fn list_occurrences(
1003        &self,
1004        filter: TriggerOccurrenceFilter,
1005    ) -> Result<Vec<TriggerOccurrenceRecord>, PluginError>;
1006
1007    async fn list_deliveries_by_occurrence_id(
1008        &self,
1009        occurrence_id: &str,
1010    ) -> Result<Vec<TriggerDeliveryReservation>, PluginError>;
1011
1012    async fn list_deliveries_by_subscription_id(
1013        &self,
1014        subscription_id: &str,
1015    ) -> Result<Vec<TriggerDeliveryReservation>, PluginError>;
1016
1017    async fn list_deliveries_by_process_id(
1018        &self,
1019        process_id: &str,
1020    ) -> Result<Vec<TriggerDeliveryReservation>, PluginError>;
1021
1022    /// List every reserved delivery snapshot, including deliveries whose live
1023    /// subscription has since been updated or tombstoned. Recovery uses this
1024    /// direct delivery-table view to close the reserve/start crash window.
1025    async fn list_deliveries(&self) -> Result<Vec<TriggerDeliveryReservation>, PluginError>;
1026
1027    /// Drop mutation idempotency receipts older than the host's established
1028    /// terminal-process retention cutoff. List operations never create these
1029    /// receipts.
1030    async fn prune_mutation_receipts(&self, cutoff_epoch_ms: u64) -> Result<usize, PluginError>;
1031}
lash_core/triggers.rs

lash_core/
triggers.rs
