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 value: String,
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, s.record_json
94 FROM trigger_deliveries d
95 JOIN trigger_occurrences o ON o.occurrence_id = d.occurrence_id
96 JOIN trigger_subscriptions s ON s.subscription_id = d.subscription_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(params![value.as_str()], |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 register_subscription(
147 &self,
148 draft: lash_core::TriggerSubscriptionDraft,
149 ) -> Result<lash_core::TriggerSubscriptionRecord, lash_core::PluginError> {
150 draft.validate()?;
151 let now = self.clock.timestamp_ms();
152 self.conn
153 .write_flow(move |tx| {
154 Ok(trigger_tx_outcome((|| {
155 tx.execute("INSERT INTO trigger_subscription_seq DEFAULT VALUES", [])
156 .map_err(process_sqlite_error)?;
157 let seq = tx.last_insert_rowid();
158 let handle = format!("trigger:{seq}");
159 let subscription_id = format!("subscription:{seq}");
160 let record = lash_core::TriggerSubscriptionRecord {
161 subscription_id: subscription_id.clone(),
162 registrant: draft.registrant,
163 env_ref: draft.env_ref,
164 wake_target: draft.wake_target,
165 handle,
166 name: draft.name,
167 source_type: draft.source_type,
168 source_key: draft.source_key,
169 source: draft.source,
170 payload_schema: draft.payload_schema,
171 target: draft.target,
172 target_identity: draft.target_identity,
173 event_types: draft.event_types,
174 input_template: draft.input_template,
175 target_label: draft.target_label,
176 enabled: true,
177 created_at_ms: now,
178 updated_at_ms: now,
179 };
180 tx.execute(
181 "INSERT INTO trigger_subscriptions (
182 subscription_id, registrant_scope_id, handle, source_type, source_key,
183 enabled, created_at_ms, updated_at_ms, record_json
184 )
185 VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9)",
186 params![
187 record.subscription_id.as_str(),
188 record.registrant_scope_id().as_str(),
189 record.handle.as_str(),
190 record.source_type.as_str(),
191 record.source_key.as_str(),
192 i64::from(record.enabled),
193 record.created_at_ms as i64,
194 record.updated_at_ms as i64,
195 Self::encode_json(&record)?,
196 ],
197 )
198 .map_err(process_sqlite_error)?;
199 Ok(record)
200 })()))
201 })
202 .await
203 .map_err(process_sqlite_error)?
204 }
205
206 async fn list_subscriptions(
207 &self,
208 filter: lash_core::TriggerSubscriptionFilter,
209 ) -> Result<Vec<lash_core::TriggerSubscriptionRecord>, lash_core::PluginError> {
210 self.conn
211 .call(move |conn| {
212 Ok((|| {
213 let mut sql =
214 "SELECT subscription_id, record_json FROM trigger_subscriptions WHERE 1 = 1"
215 .to_string();
216 let mut values = Vec::<rusqlite::types::Value>::new();
217 if let Some(registrant_scope_id) = filter.effective_registrant_scope_id() {
218 sql.push_str(" AND registrant_scope_id = ?");
219 values.push(registrant_scope_id.into());
220 }
221 if let Some(handle) = filter.handle.as_ref() {
222 sql.push_str(" AND handle = ?");
223 values.push(handle.clone().into());
224 }
225 if let Some(source_type) = filter.source_type.as_ref() {
226 sql.push_str(" AND source_type = ?");
227 values.push(source_type.clone().into());
228 }
229 if let Some(source_key) = filter.source_key.as_ref() {
230 sql.push_str(" AND source_key = ?");
231 values.push(source_key.clone().into());
232 }
233 if let Some(enabled) = filter.enabled {
234 sql.push_str(" AND enabled = ?");
235 values.push(i64::from(enabled).into());
236 }
237 sql.push_str(" ORDER BY registrant_scope_id ASC, handle ASC");
238 let mut stmt = conn.prepare(&sql).map_err(process_sqlite_error)?;
239 let rows = stmt
240 .query_map(rusqlite::params_from_iter(values.iter()), |row| {
241 Ok((row.get::<_, String>(0)?, row.get::<_, String>(1)?))
242 })
243 .map_err(process_sqlite_error)?;
244 let mut records = Vec::new();
245 for row in rows {
246 let (subscription_id, json) = row.map_err(process_sqlite_error)?;
247 let record = match Self::decode_subscription(json) {
248 Ok(record) => record,
249 Err(err) => {
250 tracing::warn!(
251 error = %err,
252 subscription_id,
253 "skipping malformed trigger subscription during listing"
254 );
255 continue;
256 }
257 };
258 if filter.matches(&record) {
259 records.push(record);
260 }
261 }
262 Ok(records)
263 })())
264 })
265 .await
266 .map_err(process_sqlite_error)?
267 }
268
269 async fn cancel_subscription(
270 &self,
271 registrant_scope_id: &str,
272 handle: &str,
273 ) -> Result<bool, lash_core::PluginError> {
274 self.set_subscription_enabled(registrant_scope_id, handle, false)
275 .await
276 }
277
278 async fn set_subscription_enabled(
279 &self,
280 registrant_scope_id: &str,
281 handle: &str,
282 enabled: bool,
283 ) -> Result<bool, lash_core::PluginError> {
284 let registrant_scope_id = registrant_scope_id.to_string();
285 let handle = handle.to_string();
286 let updated_at_ms = self.clock.timestamp_ms();
287 self.conn
288 .write_flow(move |tx| {
289 Ok(trigger_tx_outcome((|| {
290 let selected: Option<(String, i64, String)> = tx
291 .query_row(
292 "SELECT subscription_id, enabled, record_json
293 FROM trigger_subscriptions
294 WHERE registrant_scope_id = ?1 AND handle = ?2",
295 params![registrant_scope_id.as_str(), handle.as_str()],
296 |row| Ok((row.get(0)?, row.get(1)?, row.get(2)?)),
297 )
298 .optional()
299 .map_err(process_sqlite_error)?;
300 let Some((subscription_id, stored_enabled, json)) = selected else {
301 return Ok(false);
302 };
303 let changed = (stored_enabled != 0) != enabled;
304 match Self::decode_subscription(json) {
305 Ok(mut record) => {
306 record.enabled = enabled;
307 record.updated_at_ms = updated_at_ms;
308 tx.execute(
309 "UPDATE trigger_subscriptions
310 SET enabled = ?3, updated_at_ms = ?4, record_json = ?5
311 WHERE subscription_id = ?1 AND handle = ?2",
312 params![
313 subscription_id.as_str(),
314 handle.as_str(),
315 i64::from(record.enabled),
316 record.updated_at_ms as i64,
317 Self::encode_json(&record)?,
318 ],
319 )
320 .map_err(process_sqlite_error)?;
321 }
322 Err(err) => {
323 tracing::warn!(
324 error = %err,
325 subscription_id,
326 handle,
327 "disabling malformed trigger subscription without rewriting record JSON"
328 );
329 tx.execute(
330 "UPDATE trigger_subscriptions
331 SET enabled = ?3, updated_at_ms = ?4
332 WHERE subscription_id = ?1 AND handle = ?2",
333 params![
334 subscription_id.as_str(),
335 handle.as_str(),
336 i64::from(enabled),
337 updated_at_ms as i64,
338 ],
339 )
340 .map_err(process_sqlite_error)?;
341 }
342 }
343 Ok(changed)
344 })()))
345 })
346 .await
347 .map_err(process_sqlite_error)?
348 }
349
350 async fn delete_subscription(
351 &self,
352 registrant_scope_id: &str,
353 handle: &str,
354 ) -> Result<bool, lash_core::PluginError> {
355 let registrant_scope_id = registrant_scope_id.to_string();
356 let handle = handle.to_string();
357 self.conn
358 .write_flow(move |tx| {
359 Ok(trigger_tx_outcome((|| {
360 let deleted = tx
361 .execute(
362 "DELETE FROM trigger_subscriptions
363 WHERE registrant_scope_id = ?1 AND handle = ?2",
364 params![registrant_scope_id.as_str(), handle.as_str()],
365 )
366 .map_err(process_sqlite_error)?;
367 Ok(deleted != 0)
368 })()))
369 })
370 .await
371 .map_err(process_sqlite_error)?
372 }
373
374 async fn delete_session_subscriptions(
375 &self,
376 session_id: &str,
377 ) -> Result<usize, lash_core::PluginError> {
378 let session_id = session_id.to_string();
379 self.conn
380 .write_flow(move |tx| {
381 Ok(trigger_tx_outcome((|| {
382 let mut stmt = tx
383 .prepare("SELECT subscription_id, record_json FROM trigger_subscriptions")
384 .map_err(process_sqlite_error)?;
385 let rows = stmt
386 .query_map([], |row| {
387 Ok((row.get::<_, String>(0)?, row.get::<_, String>(1)?))
388 })
389 .map_err(process_sqlite_error)?;
390 let mut subscription_ids = Vec::new();
391 for row in rows {
392 let (subscription_id, json) = row.map_err(process_sqlite_error)?;
393 let record = match Self::decode_subscription(json) {
394 Ok(record) => record,
395 Err(err) => {
396 tracing::warn!(
397 error = %err,
398 subscription_id,
399 "skipping malformed trigger subscription during session delete"
400 );
401 continue;
402 }
403 };
404 if record.registrant_session_id() == Some(session_id.as_str()) {
405 subscription_ids.push(subscription_id);
406 }
407 }
408 drop(stmt);
409 let mut deleted = 0usize;
410 for subscription_id in subscription_ids {
411 deleted = deleted.saturating_add(
412 tx.execute(
413 "DELETE FROM trigger_subscriptions WHERE subscription_id = ?1",
414 params![subscription_id.as_str()],
415 )
416 .map_err(process_sqlite_error)?,
417 );
418 }
419 Ok(deleted)
420 })()))
421 })
422 .await
423 .map_err(process_sqlite_error)?
424 }
425
426 async fn record_occurrence(
427 &self,
428 request: lash_core::TriggerOccurrenceRequest,
429 ) -> Result<lash_core::TriggerOccurrenceRecord, lash_core::PluginError> {
430 lash_core::validate_trigger_occurrence_request(&request)?;
431 let request_hash = lash_core::trigger_occurrence_request_hash(&request)?;
432 let occurrence_id = lash_core::deterministic_occurrence_id(&request)?;
433 let occurred_at_ms = self.clock.timestamp_ms();
434 self.conn
435 .write_flow(move |tx| {
436 Ok(trigger_tx_outcome((|| {
437 let existing: Option<(String, String)> = tx
438 .query_row(
439 "SELECT request_hash, record_json
440 FROM trigger_occurrences
441 WHERE idempotency_key = ?1",
442 params![request.idempotency_key.as_str()],
443 |row| Ok((row.get(0)?, row.get(1)?)),
444 )
445 .optional()
446 .map_err(process_sqlite_error)?;
447 if let Some((existing_hash, existing_json)) = existing {
448 if existing_hash != request_hash {
449 return Err(lash_core::PluginError::Session(format!(
450 "trigger occurrence idempotency conflict for `{}`",
451 request.idempotency_key
452 )));
453 }
454 return Self::decode_occurrence(existing_json);
455 }
456 let record = lash_core::TriggerOccurrenceRecord {
457 occurrence_id: occurrence_id.clone(),
458 source_type: request.source_type,
459 source_key: request.source_key,
460 payload: request.payload,
461 idempotency_key: request.idempotency_key,
462 source: request.source,
463 session_id: request.session_id,
464 occurred_at_ms,
465 };
466 tx.execute(
467 "INSERT INTO trigger_occurrences (
468 occurrence_id, idempotency_key, request_hash, source_type,
469 source_key, occurred_at_ms, record_json
470 )
471 VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7)",
472 params![
473 record.occurrence_id.as_str(),
474 record.idempotency_key.as_str(),
475 request_hash.as_str(),
476 record.source_type.as_str(),
477 record.source_key.as_str(),
478 record.occurred_at_ms as i64,
479 Self::encode_json(&record)?,
480 ],
481 )
482 .map_err(process_sqlite_error)?;
483 Ok(record)
484 })()))
485 })
486 .await
487 .map_err(process_sqlite_error)?
488 }
489
490 async fn list_occurrences(
491 &self,
492 filter: lash_core::TriggerOccurrenceFilter,
493 ) -> Result<Vec<lash_core::TriggerOccurrenceRecord>, lash_core::PluginError> {
494 self.conn
495 .call(move |conn| {
496 Ok((|| {
497 let mut sql =
498 "SELECT occurrence_id, record_json FROM trigger_occurrences WHERE 1 = 1"
499 .to_string();
500 let mut values = Vec::<rusqlite::types::Value>::new();
501 if let Some(source_type) = filter.source_type.as_ref() {
502 sql.push_str(" AND source_type = ?");
503 values.push(source_type.clone().into());
504 }
505 if let Some(source_key) = filter.source_key.as_ref() {
506 sql.push_str(" AND source_key = ?");
507 values.push(source_key.clone().into());
508 }
509 if let Some(start_ms) = filter.occurred_at_start_ms {
510 sql.push_str(" AND occurred_at_ms >= ?");
511 values.push((start_ms as i64).into());
512 }
513 if let Some(end_ms) = filter.occurred_at_end_ms {
514 sql.push_str(" AND occurred_at_ms < ?");
515 values.push((end_ms as i64).into());
516 }
517 sql.push_str(" ORDER BY occurred_at_ms ASC, occurrence_id ASC");
518 let mut stmt = conn.prepare(&sql).map_err(process_sqlite_error)?;
519 let rows = stmt
520 .query_map(rusqlite::params_from_iter(values.iter()), |row| {
521 Ok((row.get::<_, String>(0)?, row.get::<_, String>(1)?))
522 })
523 .map_err(process_sqlite_error)?;
524 let mut records = Vec::new();
525 for row in rows {
526 let (occurrence_id, json) = row.map_err(process_sqlite_error)?;
527 match Self::decode_occurrence(json) {
528 Ok(record) => records.push(record),
529 Err(err) => tracing::warn!(
530 error = %err,
531 occurrence_id,
532 "skipping malformed trigger occurrence during listing"
533 ),
534 }
535 }
536 Ok(records)
537 })())
538 })
539 .await
540 .map_err(process_sqlite_error)?
541 }
542
543 async fn reserve_matching_deliveries(
544 &self,
545 occurrence_id: &str,
546 ) -> Result<Vec<lash_core::TriggerDeliveryReservation>, lash_core::PluginError> {
547 let occurrence_id = occurrence_id.to_string();
548 let created_at_ms = self.clock.timestamp_ms();
549 self.conn
550 .write_flow(move |tx| {
551 Ok(trigger_tx_outcome((|| {
552 let occurrence_json: Option<String> = tx
553 .query_row(
554 "SELECT record_json
555 FROM trigger_occurrences
556 WHERE occurrence_id = ?1",
557 params![occurrence_id.as_str()],
558 |row| row.get(0),
559 )
560 .optional()
561 .map_err(process_sqlite_error)?;
562 let Some(occurrence_json) = occurrence_json else {
563 return Err(lash_core::PluginError::Session(format!(
564 "unknown trigger occurrence `{occurrence_id}`"
565 )));
566 };
567 let occurrence = Self::decode_occurrence(occurrence_json)?;
568 let subscriptions = {
569 let mut stmt = tx
570 .prepare(
571 "SELECT subscription_id, record_json
572 FROM trigger_subscriptions
573 WHERE enabled = 1 AND source_type = ?1 AND source_key = ?2
574 ORDER BY registrant_scope_id ASC, handle ASC",
575 )
576 .map_err(process_sqlite_error)?;
577 let rows = stmt
578 .query_map(
579 params![
580 occurrence.source_type.as_str(),
581 occurrence.source_key.as_str()
582 ],
583 |row| {
584 Ok((row.get::<_, String>(0)?, row.get::<_, String>(1)?))
585 },
586 )
587 .map_err(process_sqlite_error)?;
588 let mut subscriptions = Vec::new();
589 for row in rows {
590 let (subscription_id, json) = row.map_err(process_sqlite_error)?;
591 match Self::decode_subscription(json) {
592 Ok(subscription)
593 if occurrence.session_id.as_deref().is_none_or(|session_id| {
594 subscription.registrant_session_id() == Some(session_id)
595 }) =>
596 {
597 subscriptions.push(subscription);
598 }
599 Ok(_) => {}
600 Err(err) => tracing::warn!(
601 error = %err,
602 subscription_id,
603 occurrence_id = %occurrence.occurrence_id,
604 "skipping malformed trigger subscription during delivery reservation"
605 ),
606 }
607 }
608 subscriptions
609 };
610 let mut reservations = Vec::new();
611 for subscription in subscriptions {
612 let process_id = lash_core::deterministic_delivery_process_id(
613 &occurrence.occurrence_id,
614 &subscription.subscription_id,
615 )?;
616 let inserted = tx
617 .execute(
618 "INSERT OR IGNORE INTO trigger_deliveries (
619 occurrence_id, subscription_id, process_id, created_at_ms
620 )
621 VALUES (?1, ?2, ?3, ?4)",
622 params![
623 occurrence.occurrence_id.as_str(),
624 subscription.subscription_id.as_str(),
625 process_id.as_str(),
626 created_at_ms as i64,
627 ],
628 )
629 .map_err(process_sqlite_error)?;
630 let stored_created_at_ms: i64 = tx
631 .query_row(
632 "SELECT created_at_ms FROM trigger_deliveries
633 WHERE occurrence_id = ?1 AND subscription_id = ?2",
634 params![
635 occurrence.occurrence_id.as_str(),
636 subscription.subscription_id.as_str()
637 ],
638 |row| row.get(0),
639 )
640 .map_err(process_sqlite_error)?;
641 reservations.push(lash_core::TriggerDeliveryReservation {
642 occurrence: occurrence.clone(),
643 subscription,
644 process_id,
645 created_at_ms: stored_created_at_ms as u64,
646 reservation_status: if inserted == 0 {
647 lash_core::TriggerDeliveryReservationStatus::AlreadyReserved
648 } else {
649 lash_core::TriggerDeliveryReservationStatus::Reserved
650 },
651 });
652 }
653 Ok(reservations)
654 })()))
655 })
656 .await
657 .map_err(process_sqlite_error)?
658 }
659
660 async fn list_deliveries_by_occurrence_id(
661 &self,
662 occurrence_id: &str,
663 ) -> Result<Vec<lash_core::TriggerDeliveryReservation>, lash_core::PluginError> {
664 self.list_deliveries_where("d.occurrence_id = ?1", occurrence_id.to_string())
665 .await
666 }
667
668 async fn list_deliveries_by_subscription_id(
669 &self,
670 subscription_id: &str,
671 ) -> Result<Vec<lash_core::TriggerDeliveryReservation>, lash_core::PluginError> {
672 self.list_deliveries_where("d.subscription_id = ?1", subscription_id.to_string())
673 .await
674 }
675
676 async fn list_deliveries_by_process_id(
677 &self,
678 process_id: &str,
679 ) -> Result<Vec<lash_core::TriggerDeliveryReservation>, lash_core::PluginError> {
680 self.list_deliveries_where("d.process_id = ?1", process_id.to_string())
681 .await
682 }
683}