1use super::*;
8
9pub struct SqliteTriggerStore {
10 conn: SqliteConnection,
11 clock: Arc<dyn lash_core::Clock>,
12}
13
14impl SqliteTriggerStore {
15 pub async fn open(path: &Path) -> tokio_rusqlite::Result<Self> {
16 Self::open_with_clock(path, Arc::new(lash_core::SystemClock)).await
17 }
18
19 pub async fn open_with_clock(
20 path: &Path,
21 clock: Arc<dyn lash_core::Clock>,
22 ) -> tokio_rusqlite::Result<Self> {
23 let conn = SqliteConnection::open(path).await?;
24 ensure_trigger_schema(&conn).await?;
25 apply_pragmas(&conn, StoreBacking::File).await?;
26 Ok(Self { conn, clock })
27 }
28
29 pub async fn memory() -> tokio_rusqlite::Result<Self> {
30 Self::memory_with_clock(Arc::new(lash_core::SystemClock)).await
31 }
32
33 pub async fn memory_with_clock(
34 clock: Arc<dyn lash_core::Clock>,
35 ) -> tokio_rusqlite::Result<Self> {
36 let conn = SqliteConnection::open_in_memory().await?;
37 ensure_trigger_schema(&conn).await?;
38 apply_pragmas(&conn, StoreBacking::Memory).await?;
39 Ok(Self { conn, clock })
40 }
41
42 fn encode_json<T: serde::Serialize>(value: &T) -> Result<String, lash_core::PluginError> {
43 serde_json::to_string(value).map_err(|err| {
44 lash_core::PluginError::Session(format!("failed to encode trigger row: {err}"))
45 })
46 }
47
48 fn decode_subscription(
49 json: String,
50 ) -> Result<lash_core::TriggerSubscriptionRecord, lash_core::PluginError> {
51 serde_json::from_str(&json).map_err(|err| {
52 lash_core::PluginError::Session(format!(
53 "failed to decode trigger subscription row: {err}"
54 ))
55 })
56 }
57
58 fn decode_occurrence(
59 json: String,
60 ) -> Result<lash_core::TriggerOccurrenceRecord, lash_core::PluginError> {
61 serde_json::from_str(&json).map_err(|err| {
62 lash_core::PluginError::Session(format!(
63 "failed to decode trigger occurrence row: {err}"
64 ))
65 })
66 }
67
68 fn decode_delivery(
69 occurrence_json: String,
70 subscription_json: String,
71 process_id: String,
72 created_at_ms: i64,
73 reservation_status: lash_core::TriggerDeliveryReservationStatus,
74 ) -> Result<lash_core::TriggerDeliveryReservation, lash_core::PluginError> {
75 Ok(lash_core::TriggerDeliveryReservation {
76 occurrence: Self::decode_occurrence(occurrence_json)?,
77 subscription: Self::decode_subscription(subscription_json)?,
78 process_id,
79 created_at_ms: created_at_ms as u64,
80 reservation_status,
81 })
82 }
83
84 async fn list_deliveries_where(
85 &self,
86 where_clause: &'static str,
87 values: Vec<rusqlite::types::Value>,
88 ) -> Result<Vec<lash_core::TriggerDeliveryReservation>, lash_core::PluginError> {
89 self.conn
90 .call(move |conn| {
91 Ok((|| {
92 let sql = format!(
93 "SELECT d.process_id, d.created_at_ms, o.record_json,
94 d.subscription_snapshot_json
95 FROM trigger_deliveries d
96 JOIN trigger_occurrences o ON o.occurrence_id = d.occurrence_id
97 WHERE {where_clause}
98 ORDER BY d.created_at_ms ASC, d.occurrence_id ASC, d.subscription_id ASC"
99 );
100 let mut stmt = conn.prepare(&sql).map_err(process_sqlite_error)?;
101 let rows = stmt
102 .query_map(rusqlite::params_from_iter(values.iter()), |row| {
103 Ok((
104 row.get::<_, String>(0)?,
105 row.get::<_, i64>(1)?,
106 row.get::<_, String>(2)?,
107 row.get::<_, String>(3)?,
108 ))
109 })
110 .map_err(process_sqlite_error)?;
111 let mut deliveries = Vec::new();
112 for row in rows {
113 let (process_id, created_at_ms, occurrence_json, subscription_json) =
114 row.map_err(process_sqlite_error)?;
115 deliveries.push(Self::decode_delivery(
116 occurrence_json,
117 subscription_json,
118 process_id,
119 created_at_ms,
120 lash_core::TriggerDeliveryReservationStatus::AlreadyReserved,
121 )?);
122 }
123 Ok(deliveries)
124 })())
125 })
126 .await
127 .map_err(process_sqlite_error)?
128 }
129}
130
131fn trigger_tx_outcome<T>(
132 result: Result<T, lash_core::PluginError>,
133) -> TxOutcome<Result<T, lash_core::PluginError>> {
134 match result {
135 Ok(value) => TxOutcome::Commit(Ok(value)),
136 Err(err) => TxOutcome::Rollback(Err(err)),
137 }
138}
139
140#[async_trait::async_trait]
141impl lash_core::TriggerStore for SqliteTriggerStore {
142 fn durability_tier(&self) -> DurabilityTier {
143 DurabilityTier::Durable
144 }
145
146 async fn execute_command(
147 &self,
148 operation_id: &str,
149 command: lash_core::TriggerCommand,
150 ) -> Result<lash_core::TriggerEffectResult, lash_core::PluginError> {
151 if let lash_core::TriggerCommand::List {
152 owner_scope,
153 mut filter,
154 } = command
155 {
156 filter.registrant_scope_id = Some(owner_scope.namespace());
157 return self
158 .list_subscriptions(filter)
159 .await
160 .map(|records| Ok(lash_core::TriggerCommandOutcome::List { records }));
161 }
162 let public_operation_id = operation_id.to_string();
163 let operation_id =
164 lash_core::trigger_operation_receipt_id(command.owner_scope(), operation_id)?;
165 let request_hash = lash_core::trigger_command_hash(&command)?;
166 let owner_scope = command.owner_scope().clone();
167 let subscription_key = command.subscription_key().unwrap_or_default().to_string();
168 let subscription_id =
169 lash_core::deterministic_subscription_id(&owner_scope, &subscription_key)?;
170 let now = self.clock.timestamp_ms();
171 self.conn
172 .write_flow(move |tx| {
173 Ok(trigger_tx_outcome((|| {
174 let receipt: Option<(String, String)> = tx
175 .query_row(
176 "SELECT request_hash, result_json
177 FROM trigger_mutation_receipts WHERE operation_id = ?1",
178 params![operation_id.as_str()],
179 |row| Ok((row.get(0)?, row.get(1)?)),
180 )
181 .optional()
182 .map_err(process_sqlite_error)?;
183 if let Some((stored_hash, json)) = receipt {
184 if stored_hash != request_hash {
185 return Ok(Err(lash_core::TriggerOperationError::Conflict {
186 subscription_key,
187 existing_revision: None,
188 existing_definition_hash: Some(stored_hash),
189 requested_definition_hash: Some(request_hash),
190 reason: format!(
191 "operation id `{public_operation_id}` was reused with different content"
192 ),
193 }));
194 }
195 return serde_json::from_str(&json).map_err(|err| {
196 lash_core::PluginError::Session(format!(
197 "failed to decode trigger mutation receipt: {err}"
198 ))
199 });
200 }
201 let result = if let lash_core::TriggerCommand::Prune {
202 owner_scope,
203 actor,
204 subscription_keys,
205 } = &command
206 {
207 let mut stmt = tx
208 .prepare(
209 "SELECT record_json FROM trigger_subscriptions
210 WHERE owner_scope = ?1 AND tombstoned = 0",
211 )
212 .map_err(process_sqlite_error)?;
213 let rows = stmt
214 .query_map(params![owner_scope.namespace()], |row| {
215 row.get::<_, String>(0)
216 })
217 .map_err(process_sqlite_error)?;
218 let mut records = Vec::new();
219 for row in rows {
220 records.push(Self::decode_subscription(
221 row.map_err(process_sqlite_error)?,
222 )?);
223 }
224 drop(stmt);
225 lash_core::evaluate_trigger_prune(
226 records,
227 owner_scope.clone(),
228 actor.clone(),
229 subscription_keys.clone(),
230 now,
231 )
232 } else {
233 let current = tx
234 .query_row(
235 "SELECT record_json FROM trigger_subscriptions
236 WHERE subscription_id = ?1",
237 params![subscription_id.as_str()],
238 |row| row.get::<_, String>(0),
239 )
240 .optional()
241 .map_err(process_sqlite_error)?
242 .map(Self::decode_subscription)
243 .transpose()?;
244 lash_core::evaluate_trigger_mutation(current, command, now)?
245 };
246 let records = match &result {
247 Ok(lash_core::TriggerCommandOutcome::Mutation { receipt }) => {
248 vec![&receipt.record_snapshot]
249 }
250 Ok(lash_core::TriggerCommandOutcome::Prune { receipts }) => receipts
251 .iter()
252 .map(|receipt| &receipt.record_snapshot)
253 .collect(),
254 Ok(lash_core::TriggerCommandOutcome::List { .. }) | Err(_) => Vec::new(),
255 };
256 for record in records {
257 tx.execute(
258 "INSERT INTO trigger_subscriptions (
259 subscription_id, owner_scope, subscription_key, incarnation, revision,
260 definition_hash, source_type, source_key, enabled, tombstoned,
261 created_at_ms, updated_at_ms, record_json
262 )
263 VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11, ?12, ?13)
264 ON CONFLICT(subscription_id) DO UPDATE SET
265 owner_scope = excluded.owner_scope,
266 subscription_key = excluded.subscription_key,
267 incarnation = excluded.incarnation,
268 revision = excluded.revision,
269 definition_hash = excluded.definition_hash,
270 source_type = excluded.source_type,
271 source_key = excluded.source_key,
272 enabled = excluded.enabled,
273 tombstoned = excluded.tombstoned,
274 updated_at_ms = excluded.updated_at_ms,
275 record_json = excluded.record_json",
276 params![
277 record.subscription_id.as_str(),
278 record.owner_scope.namespace(),
279 record.subscription_key.as_str(),
280 record.incarnation.as_str(),
281 record.revision as i64,
282 record.definition_hash.as_str(),
283 record.source_type.as_str(),
284 record.source_key.as_str(),
285 i64::from(record.enabled),
286 i64::from(record.tombstoned),
287 record.created_at_ms as i64,
288 record.updated_at_ms as i64,
289 Self::encode_json(&record)?,
290 ],
291 )
292 .map_err(process_sqlite_error)?;
293 }
294 tx.execute(
295 "INSERT INTO trigger_mutation_receipts (
296 operation_id, request_hash, result_json, created_at_ms
297 ) VALUES (?1, ?2, ?3, ?4)",
298 params![
299 operation_id.as_str(),
300 request_hash.as_str(),
301 Self::encode_json(&result)?,
302 now as i64,
303 ],
304 )
305 .map_err(process_sqlite_error)?;
306 Ok(result)
307 })()))
308 })
309 .await
310 .map_err(process_sqlite_error)?
311 }
312
313 async fn list_subscriptions(
314 &self,
315 filter: lash_core::TriggerSubscriptionFilter,
316 ) -> Result<Vec<lash_core::TriggerSubscriptionRecord>, lash_core::PluginError> {
317 self.conn
318 .call(move |conn| {
319 Ok((|| {
320 let mut sql =
321 "SELECT subscription_id, record_json FROM trigger_subscriptions WHERE 1 = 1"
322 .to_string();
323 let mut values = Vec::<rusqlite::types::Value>::new();
324 if let Some(registrant_scope_id) = filter.effective_registrant_scope_id() {
325 sql.push_str(" AND owner_scope = ?");
326 values.push(registrant_scope_id.into());
327 }
328 if let Some(subscription_key) = filter.subscription_key.as_ref() {
329 sql.push_str(" AND subscription_key = ?");
330 values.push(subscription_key.clone().into());
331 }
332 if let Some(source_type) = filter.source_type.as_ref() {
333 sql.push_str(" AND source_type = ?");
334 values.push(source_type.clone().into());
335 }
336 if let Some(source_key) = filter.source_key.as_ref() {
337 sql.push_str(" AND source_key = ?");
338 values.push(source_key.clone().into());
339 }
340 if let Some(enabled) = filter.enabled {
341 sql.push_str(" AND enabled = ?");
342 values.push(i64::from(enabled).into());
343 }
344 sql.push_str(
345 " AND tombstoned = 0 ORDER BY owner_scope ASC, subscription_key ASC",
346 );
347 let mut stmt = conn.prepare(&sql).map_err(process_sqlite_error)?;
348 let rows = stmt
349 .query_map(rusqlite::params_from_iter(values.iter()), |row| {
350 Ok((row.get::<_, String>(0)?, row.get::<_, String>(1)?))
351 })
352 .map_err(process_sqlite_error)?;
353 let mut records = Vec::new();
354 for row in rows {
355 let (subscription_id, json) = row.map_err(process_sqlite_error)?;
356 let record = match Self::decode_subscription(json) {
357 Ok(record) => record,
358 Err(err) => {
359 tracing::warn!(
360 error = %err,
361 subscription_id,
362 "skipping malformed trigger subscription during listing"
363 );
364 continue;
365 }
366 };
367 if filter.matches(&record) {
368 records.push(record);
369 }
370 }
371 Ok(records)
372 })())
373 })
374 .await
375 .map_err(process_sqlite_error)?
376 }
377
378 async fn delete_session_subscriptions(
379 &self,
380 session_id: &str,
381 ) -> Result<usize, lash_core::PluginError> {
382 let session_id = session_id.to_string();
383 let now = self.clock.timestamp_ms();
384 self.conn
385 .write_flow(move |tx| {
386 Ok(trigger_tx_outcome((|| {
387 let mut stmt = tx
388 .prepare("SELECT subscription_id, record_json FROM trigger_subscriptions")
389 .map_err(process_sqlite_error)?;
390 let rows = stmt
391 .query_map([], |row| {
392 Ok((row.get::<_, String>(0)?, row.get::<_, String>(1)?))
393 })
394 .map_err(process_sqlite_error)?;
395 let mut subscriptions = Vec::new();
396 for row in rows {
397 let (subscription_id, json) = row.map_err(process_sqlite_error)?;
398 let record = match Self::decode_subscription(json) {
399 Ok(record) => record,
400 Err(err) => {
401 tracing::warn!(
402 error = %err,
403 subscription_id,
404 "skipping malformed trigger subscription during session delete"
405 );
406 continue;
407 }
408 };
409 if record.registrant_session_id() == Some(session_id.as_str())
410 && !record.tombstoned
411 {
412 subscriptions.push((subscription_id, record));
413 }
414 }
415 drop(stmt);
416 let mut deleted = 0usize;
417 for (subscription_id, mut record) in subscriptions {
418 record.enabled = false;
419 record.tombstoned = true;
420 record.deleted_at_ms = Some(now);
421 record.revision = record.revision.saturating_add(1);
422 record.updated_at_ms = now;
423 deleted = deleted.saturating_add(
424 tx.execute(
425 "UPDATE trigger_subscriptions
426 SET enabled = 0, tombstoned = 1, revision = ?2,
427 updated_at_ms = ?3, record_json = ?4
428 WHERE subscription_id = ?1",
429 params![
430 subscription_id.as_str(),
431 record.revision as i64,
432 now as i64,
433 Self::encode_json(&record)?,
434 ],
435 )
436 .map_err(process_sqlite_error)?,
437 );
438 }
439 Ok(deleted)
440 })()))
441 })
442 .await
443 .map_err(process_sqlite_error)?
444 }
445
446 async fn ingest_occurrence(
447 &self,
448 request: lash_core::TriggerOccurrenceRequest,
449 ) -> Result<lash_core::TriggerIngressResult, lash_core::PluginError> {
450 lash_core::validate_trigger_occurrence_request(&request)?;
451 let request_hash = lash_core::trigger_occurrence_request_hash(&request)?;
452 let occurrence_id = lash_core::deterministic_occurrence_id(&request)?;
453 let occurred_at_ms = self.clock.timestamp_ms();
454 self.conn
455 .write_flow(move |tx| {
456 Ok(trigger_tx_outcome((|| {
457 let existing: Option<(String, String)> = tx
458 .query_row(
459 "SELECT request_hash, record_json
460 FROM trigger_occurrences
461 WHERE idempotency_key = ?1",
462 params![request.idempotency_key.as_str()],
463 |row| Ok((row.get(0)?, row.get(1)?)),
464 )
465 .optional()
466 .map_err(process_sqlite_error)?;
467 let (record, is_new) = if let Some((existing_hash, existing_json)) = existing {
468 if existing_hash != request_hash {
469 return Err(lash_core::PluginError::Session(format!(
470 "trigger occurrence idempotency conflict for `{}`",
471 request.idempotency_key
472 )));
473 }
474 (Self::decode_occurrence(existing_json)?, false)
475 } else {
476 let record = lash_core::TriggerOccurrenceRecord {
477 occurrence_id: occurrence_id.clone(),
478 source_type: request.source_type,
479 source_key: request.source_key,
480 payload: request.payload,
481 idempotency_key: request.idempotency_key,
482 source: request.source,
483 session_id: request.session_id,
484 occurred_at_ms,
485 };
486 tx.execute(
487 "INSERT INTO trigger_occurrences (
488 occurrence_id, idempotency_key, request_hash, source_type,
489 source_key, occurred_at_ms, record_json
490 )
491 VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7)",
492 params![
493 record.occurrence_id.as_str(),
494 record.idempotency_key.as_str(),
495 request_hash.as_str(),
496 record.source_type.as_str(),
497 record.source_key.as_str(),
498 record.occurred_at_ms as i64,
499 Self::encode_json(&record)?,
500 ],
501 )
502 .map_err(process_sqlite_error)?;
503 (record, true)
504 };
505 let reservations = if is_new {
506 reserve_sqlite_deliveries(tx, &record, occurred_at_ms)?
507 } else {
508 sqlite_delivery_snapshots(
509 tx,
510 &record,
511 lash_core::TriggerDeliveryReservationStatus::AlreadyReserved,
512 )?
513 };
514 Ok(lash_core::TriggerIngressResult {
515 occurrence: record,
516 reservations,
517 })
518 })()))
519 })
520 .await
521 .map_err(process_sqlite_error)?
522 }
523
524 async fn list_occurrences(
525 &self,
526 filter: lash_core::TriggerOccurrenceFilter,
527 ) -> Result<Vec<lash_core::TriggerOccurrenceRecord>, lash_core::PluginError> {
528 self.conn
529 .call(move |conn| {
530 Ok((|| {
531 let mut sql =
532 "SELECT occurrence_id, record_json FROM trigger_occurrences WHERE 1 = 1"
533 .to_string();
534 let mut values = Vec::<rusqlite::types::Value>::new();
535 if let Some(source_type) = filter.source_type.as_ref() {
536 sql.push_str(" AND source_type = ?");
537 values.push(source_type.clone().into());
538 }
539 if let Some(source_key) = filter.source_key.as_ref() {
540 sql.push_str(" AND source_key = ?");
541 values.push(source_key.clone().into());
542 }
543 if let Some(start_ms) = filter.occurred_at_start_ms {
544 sql.push_str(" AND occurred_at_ms >= ?");
545 values.push((start_ms as i64).into());
546 }
547 if let Some(end_ms) = filter.occurred_at_end_ms {
548 sql.push_str(" AND occurred_at_ms < ?");
549 values.push((end_ms as i64).into());
550 }
551 sql.push_str(" ORDER BY occurred_at_ms ASC, occurrence_id ASC");
552 let mut stmt = conn.prepare(&sql).map_err(process_sqlite_error)?;
553 let rows = stmt
554 .query_map(rusqlite::params_from_iter(values.iter()), |row| {
555 Ok((row.get::<_, String>(0)?, row.get::<_, String>(1)?))
556 })
557 .map_err(process_sqlite_error)?;
558 let mut records = Vec::new();
559 for row in rows {
560 let (occurrence_id, json) = row.map_err(process_sqlite_error)?;
561 match Self::decode_occurrence(json) {
562 Ok(record) => records.push(record),
563 Err(err) => tracing::warn!(
564 error = %err,
565 occurrence_id,
566 "skipping malformed trigger occurrence during listing"
567 ),
568 }
569 }
570 Ok(records)
571 })())
572 })
573 .await
574 .map_err(process_sqlite_error)?
575 }
576
577 async fn list_deliveries_by_occurrence_id(
578 &self,
579 occurrence_id: &str,
580 ) -> Result<Vec<lash_core::TriggerDeliveryReservation>, lash_core::PluginError> {
581 self.list_deliveries_where(
582 "d.occurrence_id = ?1",
583 vec![occurrence_id.to_string().into()],
584 )
585 .await
586 }
587
588 async fn list_deliveries_by_subscription_id(
589 &self,
590 subscription_id: &str,
591 ) -> Result<Vec<lash_core::TriggerDeliveryReservation>, lash_core::PluginError> {
592 self.list_deliveries_where(
593 "d.subscription_id = ?1",
594 vec![subscription_id.to_string().into()],
595 )
596 .await
597 }
598
599 async fn list_deliveries_by_process_id(
600 &self,
601 process_id: &str,
602 ) -> Result<Vec<lash_core::TriggerDeliveryReservation>, lash_core::PluginError> {
603 self.list_deliveries_where("d.process_id = ?1", vec![process_id.to_string().into()])
604 .await
605 }
606
607 async fn list_deliveries(
608 &self,
609 ) -> Result<Vec<lash_core::TriggerDeliveryReservation>, lash_core::PluginError> {
610 self.list_deliveries_where("1 = 1", Vec::new()).await
611 }
612
613 async fn prune_mutation_receipts(
614 &self,
615 cutoff_epoch_ms: u64,
616 ) -> Result<usize, lash_core::PluginError> {
617 let cutoff_epoch_ms = i64::try_from(cutoff_epoch_ms).unwrap_or(i64::MAX);
618 self.conn
619 .call(move |conn| {
620 conn.execute(
621 "DELETE FROM trigger_mutation_receipts WHERE created_at_ms < ?1",
622 params![cutoff_epoch_ms],
623 )
624 })
625 .await
626 .map_err(process_sqlite_error)
627 }
628}
629
630fn reserve_sqlite_deliveries(
631 tx: &rusqlite::Transaction<'_>,
632 occurrence: &lash_core::TriggerOccurrenceRecord,
633 created_at_ms: u64,
634) -> Result<Vec<lash_core::TriggerDeliveryReservation>, lash_core::PluginError> {
635 let mut sql = "SELECT subscription_id, record_json FROM trigger_subscriptions
636 WHERE enabled = 1 AND tombstoned = 0 AND source_type = ?1 AND source_key = ?2"
637 .to_string();
638 let mut values: Vec<rusqlite::types::Value> = vec![
639 occurrence.source_type.clone().into(),
640 occurrence.source_key.clone().into(),
641 ];
642 if let Some(session_id) = occurrence.session_id.as_deref() {
643 sql.push_str(" AND owner_scope = ?3");
644 values.push(format!("session:{session_id}").into());
645 }
646 sql.push_str(" ORDER BY owner_scope ASC, subscription_key ASC");
647 let mut stmt = tx.prepare(&sql).map_err(process_sqlite_error)?;
648 let rows = stmt
649 .query_map(rusqlite::params_from_iter(values.iter()), |row| {
650 Ok((row.get::<_, String>(0)?, row.get::<_, String>(1)?))
651 })
652 .map_err(process_sqlite_error)?;
653 let mut subscriptions = Vec::new();
654 for row in rows {
655 let (subscription_id, json) = row.map_err(process_sqlite_error)?;
656 match SqliteTriggerStore::decode_subscription(json) {
657 Ok(subscription) => subscriptions.push(subscription),
658 Err(err) => tracing::warn!(
659 error = %err,
660 subscription_id,
661 "skipping malformed trigger subscription during occurrence ingress"
662 ),
663 }
664 }
665 drop(stmt);
666
667 let mut reservations = Vec::with_capacity(subscriptions.len());
668 for subscription in subscriptions {
669 let process_id = lash_core::deterministic_delivery_process_id(
670 &occurrence.occurrence_id,
671 &subscription.subscription_id,
672 &subscription.incarnation,
673 subscription.revision,
674 )?;
675 tx.execute(
676 "INSERT INTO trigger_deliveries (
677 occurrence_id, subscription_id, process_id, subscription_incarnation,
678 subscription_revision, subscription_snapshot_json, created_at_ms
679 ) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7)",
680 params![
681 occurrence.occurrence_id.as_str(),
682 subscription.subscription_id.as_str(),
683 process_id.as_str(),
684 subscription.incarnation.as_str(),
685 subscription.revision as i64,
686 SqliteTriggerStore::encode_json(&subscription)?,
687 created_at_ms as i64,
688 ],
689 )
690 .map_err(process_sqlite_error)?;
691 reservations.push(lash_core::TriggerDeliveryReservation {
692 occurrence: occurrence.clone(),
693 subscription,
694 process_id,
695 created_at_ms,
696 reservation_status: lash_core::TriggerDeliveryReservationStatus::Reserved,
697 });
698 }
699 Ok(reservations)
700}
701
702fn sqlite_delivery_snapshots(
703 tx: &rusqlite::Transaction<'_>,
704 occurrence: &lash_core::TriggerOccurrenceRecord,
705 reservation_status: lash_core::TriggerDeliveryReservationStatus,
706) -> Result<Vec<lash_core::TriggerDeliveryReservation>, lash_core::PluginError> {
707 let mut stmt = tx
708 .prepare(
709 "SELECT process_id, created_at_ms, subscription_snapshot_json
710 FROM trigger_deliveries
711 WHERE occurrence_id = ?1
712 ORDER BY created_at_ms ASC, subscription_id ASC",
713 )
714 .map_err(process_sqlite_error)?;
715 let rows = stmt
716 .query_map(params![occurrence.occurrence_id.as_str()], |row| {
717 Ok((
718 row.get::<_, String>(0)?,
719 row.get::<_, i64>(1)?,
720 row.get::<_, String>(2)?,
721 ))
722 })
723 .map_err(process_sqlite_error)?;
724 let mut reservations = Vec::new();
725 for row in rows {
726 let (process_id, created_at_ms, snapshot_json) = row.map_err(process_sqlite_error)?;
727 reservations.push(lash_core::TriggerDeliveryReservation {
728 occurrence: occurrence.clone(),
729 subscription: SqliteTriggerStore::decode_subscription(snapshot_json)?,
730 process_id,
731 created_at_ms: created_at_ms as u64,
732 reservation_status: reservation_status.clone(),
733 });
734 }
735 Ok(reservations)
736}