1use super::*;
2
3pub struct InMemoryTriggerStore {
4 clock: Arc<dyn crate::Clock>,
5 state: Mutex<InMemoryTriggerEventState>,
6}
7
8impl InMemoryTriggerStore {
9 pub fn new() -> Self {
10 Self::with_clock(Arc::new(crate::SystemClock))
11 }
12
13 pub fn with_clock(clock: Arc<dyn crate::Clock>) -> Self {
14 Self {
15 clock,
16 state: Mutex::new(InMemoryTriggerEventState::default()),
17 }
18 }
19
20 fn list_deliveries_matching(
21 &self,
22 matches: impl Fn(&InMemoryTriggerDeliveryRecord) -> bool,
23 ) -> Result<Vec<TriggerDeliveryReservation>, PluginError> {
24 let state = self
25 .state
26 .lock()
27 .map_err(|_| PluginError::Session("trigger store lock poisoned".to_string()))?;
28 let mut deliveries = state
29 .deliveries
30 .values()
31 .filter(|delivery| matches(delivery))
32 .map(|delivery| {
33 in_memory_delivery_reservation(
34 &state,
35 delivery,
36 TriggerDeliveryReservationStatus::AlreadyReserved,
37 )
38 })
39 .collect::<Result<Vec<_>, _>>()?;
40 deliveries.sort_by(|left, right| {
41 left.created_at_ms
42 .cmp(&right.created_at_ms)
43 .then_with(|| {
44 left.occurrence
45 .occurrence_id
46 .cmp(&right.occurrence.occurrence_id)
47 })
48 .then_with(|| {
49 left.subscription
50 .subscription_id
51 .cmp(&right.subscription.subscription_id)
52 })
53 });
54 Ok(deliveries)
55 }
56
57 #[cfg(any(test, feature = "testing"))]
58 pub(crate) fn delete_deliveries_by_process_ids(
59 &self,
60 process_ids: &std::collections::HashSet<String>,
61 ) -> Result<usize, PluginError> {
62 if process_ids.is_empty() {
63 return Ok(0);
64 }
65 let mut state = self
66 .state
67 .lock()
68 .map_err(|_| PluginError::Session("trigger store lock poisoned".to_string()))?;
69 let before = state.deliveries.len();
70 state
71 .deliveries
72 .retain(|_, delivery| !process_ids.contains(&delivery.process_id));
73 Ok(before.saturating_sub(state.deliveries.len()))
74 }
75}
76
77impl Default for InMemoryTriggerStore {
78 fn default() -> Self {
79 Self::new()
80 }
81}
82
83#[derive(Default)]
84pub(super) struct InMemoryTriggerEventState {
85 pub(super) subscriptions: BTreeMap<String, TriggerSubscriptionRecord>,
86 pub(super) mutation_receipts: BTreeMap<String, (String, TriggerEffectResult, u64)>,
87 pub(super) occurrences: BTreeMap<String, TriggerOccurrenceRecord>,
88 pub(super) occurrence_id_by_idempotency_key: BTreeMap<String, String>,
89 pub(super) occurrence_hashes: BTreeMap<String, String>,
90 pub(super) deliveries: BTreeMap<(String, String), InMemoryTriggerDeliveryRecord>,
91}
92
93#[derive(Clone)]
94pub(super) struct InMemoryTriggerDeliveryRecord {
95 pub(super) occurrence_id: String,
96 pub(super) subscription_id: String,
97 pub(super) process_id: String,
98 pub(super) created_at_ms: u64,
99 pub(super) subscription_snapshot: TriggerSubscriptionRecord,
100}
101
102fn in_memory_delivery_reservation(
103 state: &InMemoryTriggerEventState,
104 delivery: &InMemoryTriggerDeliveryRecord,
105 reservation_status: TriggerDeliveryReservationStatus,
106) -> Result<TriggerDeliveryReservation, PluginError> {
107 let occurrence = state
108 .occurrences
109 .get(&delivery.occurrence_id)
110 .cloned()
111 .ok_or_else(|| {
112 PluginError::Session(format!(
113 "missing trigger occurrence `{}` for delivery",
114 delivery.occurrence_id
115 ))
116 })?;
117 let subscription = delivery.subscription_snapshot.clone();
118 Ok(TriggerDeliveryReservation {
119 occurrence,
120 subscription,
121 process_id: delivery.process_id.clone(),
122 created_at_ms: delivery.created_at_ms,
123 reservation_status,
124 })
125}
126
127#[async_trait::async_trait]
128impl TriggerStore for InMemoryTriggerStore {
129 async fn execute_command(
130 &self,
131 operation_id: &str,
132 command: TriggerCommand,
133 ) -> Result<TriggerEffectResult, PluginError> {
134 if operation_id.trim().is_empty() {
135 return Ok(Err(TriggerOperationError::Invalid {
136 message: "trigger operation id must be non-empty".to_string(),
137 }));
138 }
139 let mut state = self
140 .state
141 .lock()
142 .map_err(|_| PluginError::Session("trigger store lock poisoned".to_string()))?;
143 execute_in_memory_trigger_command(
144 &mut state,
145 operation_id,
146 command,
147 self.clock.timestamp_ms(),
148 )
149 }
150
151 async fn list_subscriptions(
152 &self,
153 filter: TriggerSubscriptionFilter,
154 ) -> Result<Vec<TriggerSubscriptionRecord>, PluginError> {
155 let state = self
156 .state
157 .lock()
158 .map_err(|_| PluginError::Session("trigger store lock poisoned".to_string()))?;
159 let mut records = state
160 .subscriptions
161 .values()
162 .filter(|record| filter.matches(record))
163 .cloned()
164 .collect::<Vec<_>>();
165 records.sort_by(|left, right| {
166 left.registrant_scope_id()
167 .cmp(&right.registrant_scope_id())
168 .then_with(|| left.subscription_key.cmp(&right.subscription_key))
169 });
170 Ok(records)
171 }
172
173 async fn delete_session_subscriptions(&self, session_id: &str) -> Result<usize, PluginError> {
174 let mut state = self
175 .state
176 .lock()
177 .map_err(|_| PluginError::Session("trigger store lock poisoned".to_string()))?;
178 let mut changed = 0usize;
179 let now = self.clock.timestamp_ms();
180 for record in state.subscriptions.values_mut().filter(|record| {
181 record.registrant_session_id() == Some(session_id) && !record.tombstoned
182 }) {
183 record.enabled = false;
184 record.tombstoned = true;
185 record.deleted_at_ms = Some(now);
186 record.revision = record.revision.saturating_add(1);
187 record.updated_at_ms = now;
188 changed = changed.saturating_add(1);
189 }
190 Ok(changed)
191 }
192
193 async fn ingest_occurrence(
194 &self,
195 request: TriggerOccurrenceRequest,
196 ) -> Result<TriggerIngressResult, PluginError> {
197 validate_trigger_occurrence_request(&request)?;
198 let mut state = self
199 .state
200 .lock()
201 .map_err(|_| PluginError::Session("trigger store lock poisoned".to_string()))?;
202 let request_hash = trigger_occurrence_request_hash(&request)?;
203 if let Some(existing_id) = state
204 .occurrence_id_by_idempotency_key
205 .get(&request.idempotency_key)
206 .cloned()
207 {
208 let existing_hash = state
209 .occurrence_hashes
210 .get(&existing_id)
211 .cloned()
212 .unwrap_or_default();
213 if existing_hash != request_hash {
214 return Err(PluginError::Session(format!(
215 "trigger occurrence idempotency conflict for `{}`",
216 request.idempotency_key
217 )));
218 }
219 let occurrence = state.occurrences.get(&existing_id).cloned().ok_or_else(|| {
220 PluginError::Session(format!(
221 "missing trigger occurrence `{existing_id}` for idempotency key"
222 ))
223 });
224 let occurrence = occurrence?;
225 let reservations = state
226 .deliveries
227 .values()
228 .filter(|delivery| delivery.occurrence_id == existing_id)
229 .map(|delivery| {
230 in_memory_delivery_reservation(
231 &state,
232 delivery,
233 TriggerDeliveryReservationStatus::AlreadyReserved,
234 )
235 })
236 .collect::<Result<Vec<_>, _>>()?;
237 return Ok(TriggerIngressResult {
238 occurrence,
239 reservations,
240 });
241 }
242 let occurrence_id = deterministic_occurrence_id(&request)?;
243 let record = TriggerOccurrenceRecord {
244 occurrence_id: occurrence_id.clone(),
245 source_type: request.source_type,
246 source_key: request.source_key,
247 payload: request.payload,
248 idempotency_key: request.idempotency_key.clone(),
249 source: request.source,
250 session_id: request.session_id,
251 occurred_at_ms: self.clock.timestamp_ms(),
252 };
253 state
254 .occurrence_id_by_idempotency_key
255 .insert(request.idempotency_key, occurrence_id.clone());
256 state
257 .occurrence_hashes
258 .insert(occurrence_id.clone(), request_hash);
259 state.occurrences.insert(occurrence_id, record.clone());
260 let reservations =
261 reserve_in_memory_for_occurrence(&mut state, &record, self.clock.as_ref())?;
262 Ok(TriggerIngressResult {
263 occurrence: record,
264 reservations,
265 })
266 }
267
268 async fn list_occurrences(
269 &self,
270 filter: TriggerOccurrenceFilter,
271 ) -> Result<Vec<TriggerOccurrenceRecord>, PluginError> {
272 let state = self
273 .state
274 .lock()
275 .map_err(|_| PluginError::Session("trigger store lock poisoned".to_string()))?;
276 let mut records = state
277 .occurrences
278 .values()
279 .filter(|record| filter.matches(record))
280 .cloned()
281 .collect::<Vec<_>>();
282 records.sort_by(|left, right| {
283 left.occurred_at_ms
284 .cmp(&right.occurred_at_ms)
285 .then_with(|| left.occurrence_id.cmp(&right.occurrence_id))
286 });
287 Ok(records)
288 }
289
290 async fn list_deliveries_by_occurrence_id(
291 &self,
292 occurrence_id: &str,
293 ) -> Result<Vec<TriggerDeliveryReservation>, PluginError> {
294 self.list_deliveries_matching(|delivery| delivery.occurrence_id == occurrence_id)
295 }
296
297 async fn list_deliveries_by_subscription_id(
298 &self,
299 subscription_id: &str,
300 ) -> Result<Vec<TriggerDeliveryReservation>, PluginError> {
301 self.list_deliveries_matching(|delivery| delivery.subscription_id == subscription_id)
302 }
303
304 async fn list_deliveries_by_process_id(
305 &self,
306 process_id: &str,
307 ) -> Result<Vec<TriggerDeliveryReservation>, PluginError> {
308 self.list_deliveries_matching(|delivery| delivery.process_id == process_id)
309 }
310
311 async fn list_deliveries(&self) -> Result<Vec<TriggerDeliveryReservation>, PluginError> {
312 self.list_deliveries_matching(|_| true)
313 }
314
315 async fn prune_mutation_receipts(&self, cutoff_epoch_ms: u64) -> Result<usize, PluginError> {
316 let mut state = self
317 .state
318 .lock()
319 .map_err(|_| PluginError::Session("trigger store lock poisoned".to_string()))?;
320 let before = state.mutation_receipts.len();
321 state
322 .mutation_receipts
323 .retain(|_, (_, _, created_at_ms)| *created_at_ms >= cutoff_epoch_ms);
324 Ok(before.saturating_sub(state.mutation_receipts.len()))
325 }
326}
327
328fn execute_in_memory_trigger_command(
329 state: &mut InMemoryTriggerEventState,
330 operation_id: &str,
331 command: TriggerCommand,
332 now: u64,
333) -> Result<TriggerEffectResult, PluginError> {
334 let is_mutation = command.is_mutation();
335 let request_hash = trigger_command_hash(&command)?;
336 let receipt_id = trigger_operation_receipt_id(command.owner_scope(), operation_id)?;
337 if is_mutation
338 && let Some((existing_hash, existing_result, _)) = state.mutation_receipts.get(&receipt_id)
339 {
340 if existing_hash == &request_hash {
341 return Ok(existing_result.clone());
342 }
343 return Ok(Err(TriggerOperationError::Conflict {
344 subscription_key: command.subscription_key().unwrap_or_default().to_string(),
345 existing_revision: None,
346 existing_definition_hash: Some(existing_hash.clone()),
347 requested_definition_hash: Some(request_hash),
348 reason: format!("operation id `{operation_id}` was reused with different content"),
349 }));
350 }
351
352 let result = apply_in_memory_trigger_command(state, command, now);
353 if is_mutation {
354 state
355 .mutation_receipts
356 .insert(receipt_id, (request_hash, result.clone(), now));
357 }
358 Ok(result)
359}
360
361pub(super) fn apply_in_memory_trigger_command(
362 state: &mut InMemoryTriggerEventState,
363 command: TriggerCommand,
364 now: u64,
365) -> TriggerEffectResult {
366 match command {
367 TriggerCommand::List {
368 owner_scope,
369 mut filter,
370 } => {
371 filter.registrant_scope_id = Some(owner_scope.namespace());
372 let mut records = state
373 .subscriptions
374 .values()
375 .filter(|record| filter.matches(record))
376 .cloned()
377 .collect::<Vec<_>>();
378 records.sort_by(|left, right| left.subscription_key.cmp(&right.subscription_key));
379 Ok(TriggerCommandOutcome::List { records })
380 }
381 TriggerCommand::Prune {
382 owner_scope,
383 actor,
384 subscription_keys,
385 } => {
386 let records = state.subscriptions.values().cloned().collect::<Vec<_>>();
387 let result =
388 evaluate_trigger_prune(records, owner_scope, actor, subscription_keys, now)?;
389 if let TriggerCommandOutcome::Prune { receipts } = &result {
390 for receipt in receipts {
391 state.subscriptions.insert(
392 receipt.subscription_id.clone(),
393 receipt.record_snapshot.clone(),
394 );
395 }
396 }
397 Ok(result)
398 }
399 TriggerCommand::Register {
400 owner_scope,
401 actor,
402 draft,
403 } => {
404 draft.validate().map_err(TriggerOperationError::from)?;
405 let subscription_id =
406 deterministic_subscription_id(&owner_scope, &draft.subscription_key)
407 .map_err(TriggerOperationError::from)?;
408 let definition_hash = trigger_subscription_definition_hash(&owner_scope, &draft)
409 .map_err(TriggerOperationError::from)?;
410 if let Some(existing) = state.subscriptions.get(&subscription_id).cloned() {
411 if !existing.tombstoned && existing.definition_hash == definition_hash {
412 return Ok(TriggerCommandOutcome::Mutation {
413 receipt: Box::new(TriggerMutationReceipt::from_record(
414 existing,
415 TriggerMutationDisposition::Unchanged,
416 )),
417 });
418 }
419 return Err(subscription_conflict(
420 &draft.subscription_key,
421 Some(&existing),
422 Some(definition_hash),
423 if existing.tombstoned {
424 "subscription is tombstoned; use revive"
425 } else {
426 "register does not replace a different definition; use update"
427 },
428 ));
429 }
430 let record = subscription_record_from_draft(
431 owner_scope,
432 actor,
433 draft,
434 subscription_id.clone(),
435 uuid::Uuid::new_v4().to_string(),
436 1,
437 definition_hash,
438 true,
439 now,
440 now,
441 );
442 state.subscriptions.insert(subscription_id, record.clone());
443 Ok(TriggerCommandOutcome::Mutation {
444 receipt: Box::new(TriggerMutationReceipt::from_record(
445 record,
446 TriggerMutationDisposition::Created,
447 )),
448 })
449 }
450 TriggerCommand::Update {
451 owner_scope,
452 actor,
453 subscription_key,
454 mut draft,
455 expected_revision,
456 } => {
457 draft.subscription_key.clone_from(&subscription_key);
458 draft.validate().map_err(TriggerOperationError::from)?;
459 let subscription_id = deterministic_subscription_id(&owner_scope, &subscription_key)
460 .map_err(TriggerOperationError::from)?;
461 let requested_hash = trigger_subscription_definition_hash(&owner_scope, &draft)
462 .map_err(TriggerOperationError::from)?;
463 let Some(existing) = state.subscriptions.get(&subscription_id).cloned() else {
464 return Err(subscription_conflict(
465 &subscription_key,
466 None,
467 Some(requested_hash),
468 "subscription does not exist",
469 ));
470 };
471 ensure_live_revision(&existing, expected_revision, Some(requested_hash.clone()))?;
472 let record = subscription_record_from_draft(
473 owner_scope,
474 actor,
475 draft,
476 subscription_id.clone(),
477 existing.incarnation,
478 existing.revision.saturating_add(1),
479 requested_hash,
480 existing.enabled,
481 existing.created_at_ms,
482 now,
483 );
484 state.subscriptions.insert(subscription_id, record.clone());
485 Ok(TriggerCommandOutcome::Mutation {
486 receipt: Box::new(TriggerMutationReceipt::from_record(
487 record,
488 TriggerMutationDisposition::Updated,
489 )),
490 })
491 }
492 TriggerCommand::Enable {
493 owner_scope,
494 actor,
495 subscription_key,
496 expected_revision,
497 } => mutate_enabled(
498 state,
499 owner_scope,
500 actor,
501 subscription_key,
502 expected_revision,
503 true,
504 now,
505 ),
506 TriggerCommand::Disable {
507 owner_scope,
508 actor,
509 subscription_key,
510 expected_revision,
511 } => mutate_enabled(
512 state,
513 owner_scope,
514 actor,
515 subscription_key,
516 expected_revision,
517 false,
518 now,
519 ),
520 TriggerCommand::Delete {
521 owner_scope,
522 actor,
523 subscription_key,
524 expected_revision,
525 } => {
526 let subscription_id = deterministic_subscription_id(&owner_scope, &subscription_key)
527 .map_err(TriggerOperationError::from)?;
528 let Some(existing) = state.subscriptions.get_mut(&subscription_id) else {
529 return Err(subscription_conflict(
530 &subscription_key,
531 None,
532 None,
533 "subscription does not exist",
534 ));
535 };
536 ensure_live_revision(existing, expected_revision, None)?;
537 existing.registrant = actor;
538 existing.enabled = false;
539 existing.tombstoned = true;
540 existing.deleted_at_ms = Some(now);
541 existing.revision = existing.revision.saturating_add(1);
542 existing.updated_at_ms = now;
543 Ok(TriggerCommandOutcome::Mutation {
544 receipt: Box::new(TriggerMutationReceipt::from_record(
545 existing.clone(),
546 TriggerMutationDisposition::Deleted,
547 )),
548 })
549 }
550 TriggerCommand::Revive {
551 owner_scope,
552 actor,
553 subscription_key,
554 mut draft,
555 expected_revision,
556 } => {
557 draft.subscription_key.clone_from(&subscription_key);
558 draft.validate().map_err(TriggerOperationError::from)?;
559 let subscription_id = deterministic_subscription_id(&owner_scope, &subscription_key)
560 .map_err(TriggerOperationError::from)?;
561 let requested_hash = trigger_subscription_definition_hash(&owner_scope, &draft)
562 .map_err(TriggerOperationError::from)?;
563 let Some(existing) = state.subscriptions.get(&subscription_id).cloned() else {
564 return Err(subscription_conflict(
565 &subscription_key,
566 None,
567 Some(requested_hash),
568 "subscription does not exist; use register",
569 ));
570 };
571 if !existing.tombstoned || existing.revision != expected_revision {
572 return Err(subscription_conflict(
573 &subscription_key,
574 Some(&existing),
575 Some(requested_hash),
576 "revive requires the current tombstone revision",
577 ));
578 }
579 let record = subscription_record_from_draft(
580 owner_scope,
581 actor,
582 draft,
583 subscription_id.clone(),
584 uuid::Uuid::new_v4().to_string(),
585 existing.revision.saturating_add(1),
586 requested_hash,
587 true,
588 existing.created_at_ms,
589 now,
590 );
591 state.subscriptions.insert(subscription_id, record.clone());
592 Ok(TriggerCommandOutcome::Mutation {
593 receipt: Box::new(TriggerMutationReceipt::from_record(
594 record,
595 TriggerMutationDisposition::Revived,
596 )),
597 })
598 }
599 }
600}