lash_core/triggers/
mutation.rs

1use super::*;
2
3/// Evaluate one mutation against the current logical row. Durable stores use
4/// this shared oracle inside their own transaction, then persist the receipt
5/// and returned record snapshot atomically.
6pub fn evaluate_trigger_mutation(
7    current: Option<TriggerSubscriptionRecord>,
8    command: TriggerCommand,
9    now: u64,
10) -> Result<TriggerEffectResult, PluginError> {
11    if !command.is_mutation() {
12        return Err(PluginError::Session(
13            "trigger mutation evaluator received a list command".to_string(),
14        ));
15    }
16    let mut state = InMemoryTriggerEventState::default();
17    if let Some(record) = current {
18        state
19            .subscriptions
20            .insert(record.subscription_id.clone(), record);
21    }
22    Ok(apply_in_memory_trigger_command(&mut state, command, now))
23}
24
25pub(super) fn mutate_enabled(
26    state: &mut InMemoryTriggerEventState,
27    owner_scope: TriggerOwnerScope,
28    actor: crate::ProcessOriginator,
29    subscription_key: String,
30    expected_revision: u64,
31    enabled: bool,
32    now: u64,
33) -> TriggerEffectResult {
34    let subscription_id = deterministic_subscription_id(&owner_scope, &subscription_key)
35        .map_err(TriggerOperationError::from)?;
36    let Some(existing) = state.subscriptions.get_mut(&subscription_id) else {
37        return Err(subscription_conflict(
38            &subscription_key,
39            None,
40            None,
41            "subscription does not exist",
42        ));
43    };
44    ensure_live_revision(existing, expected_revision, None)?;
45    if existing.enabled != enabled {
46        existing.enabled = enabled;
47        existing.registrant = actor;
48        existing.revision = existing.revision.saturating_add(1);
49        existing.updated_at_ms = now;
50    }
51    let disposition = if enabled {
52        TriggerMutationDisposition::Enabled
53    } else {
54        TriggerMutationDisposition::Disabled
55    };
56    Ok(TriggerCommandOutcome::Mutation {
57        receipt: Box::new(TriggerMutationReceipt::from_record(
58            existing.clone(),
59            disposition,
60        )),
61    })
62}
63
64pub(super) fn ensure_live_revision(
65    existing: &TriggerSubscriptionRecord,
66    expected_revision: u64,
67    requested_hash: Option<String>,
68) -> Result<(), TriggerOperationError> {
69    if existing.tombstoned || existing.revision != expected_revision {
70        return Err(subscription_conflict(
71            &existing.subscription_key,
72            Some(existing),
73            requested_hash,
74            if existing.tombstoned {
75                "subscription is tombstoned"
76            } else {
77                "expected revision does not match"
78            },
79        ));
80    }
81    Ok(())
82}
83
84pub(super) fn subscription_conflict(
85    subscription_key: &str,
86    existing: Option<&TriggerSubscriptionRecord>,
87    requested_definition_hash: Option<String>,
88    reason: &str,
89) -> TriggerOperationError {
90    TriggerOperationError::Conflict {
91        subscription_key: subscription_key.to_string(),
92        existing_revision: existing.map(|record| record.revision),
93        existing_definition_hash: existing.map(|record| record.definition_hash.clone()),
94        requested_definition_hash,
95        reason: reason.to_string(),
96    }
97}
98
99#[allow(clippy::too_many_arguments)]
100pub(super) fn subscription_record_from_draft(
101    owner_scope: TriggerOwnerScope,
102    actor: crate::ProcessOriginator,
103    draft: TriggerSubscriptionDraft,
104    subscription_id: String,
105    incarnation: String,
106    revision: u64,
107    definition_hash: String,
108    enabled: bool,
109    created_at_ms: u64,
110    updated_at_ms: u64,
111) -> TriggerSubscriptionRecord {
112    TriggerSubscriptionRecord {
113        subscription_id,
114        owner_scope,
115        subscription_key: draft.subscription_key,
116        incarnation,
117        revision,
118        definition_hash,
119        registrant: actor,
120        env_ref: draft.env_ref,
121        wake_target: draft.wake_target,
122        name: draft.name,
123        source_type: draft.source_type,
124        source_key: draft.source_key,
125        source: draft.source,
126        payload_schema: draft.payload_schema,
127        target: draft.target,
128        target_identity: draft.target_identity,
129        event_types: draft.event_types,
130        input_template: draft.input_template,
131        target_label: draft.target_label,
132        enabled,
133        tombstoned: false,
134        deleted_at_ms: None,
135        created_at_ms,
136        updated_at_ms,
137    }
138}
lash_core/triggers/mutation.rs

lash_core/triggers/
mutation.rs
