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 let registrant_scope_id = registrant_scope_id.to_string();
275 let handle = handle.to_string();
276 let updated_at_ms = self.clock.timestamp_ms();
277 self.conn
278 .write_flow(move |tx| {
279 Ok(trigger_tx_outcome((|| {
280 let selected: Option<(String, i64, String)> = tx
281 .query_row(
282 "SELECT subscription_id, enabled, record_json
283 FROM trigger_subscriptions
284 WHERE registrant_scope_id = ?1 AND handle = ?2",
285 params![registrant_scope_id.as_str(), handle.as_str()],
286 |row| Ok((row.get(0)?, row.get(1)?, row.get(2)?)),
287 )
288 .optional()
289 .map_err(process_sqlite_error)?;
290 let Some((subscription_id, enabled, json)) = selected else {
291 return Ok(false);
292 };
293 let changed = enabled != 0;
294 match Self::decode_subscription(json) {
295 Ok(mut record) => {
296 record.enabled = false;
297 record.updated_at_ms = updated_at_ms;
298 tx.execute(
299 "UPDATE trigger_subscriptions
300 SET enabled = ?3, updated_at_ms = ?4, record_json = ?5
301 WHERE subscription_id = ?1 AND handle = ?2",
302 params![
303 subscription_id.as_str(),
304 handle.as_str(),
305 i64::from(record.enabled),
306 record.updated_at_ms as i64,
307 Self::encode_json(&record)?,
308 ],
309 )
310 .map_err(process_sqlite_error)?;
311 }
312 Err(err) => {
313 tracing::warn!(
314 error = %err,
315 subscription_id,
316 handle,
317 "disabling malformed trigger subscription without rewriting record JSON"
318 );
319 tx.execute(
320 "UPDATE trigger_subscriptions
321 SET enabled = ?3, updated_at_ms = ?4
322 WHERE subscription_id = ?1 AND handle = ?2",
323 params![
324 subscription_id.as_str(),
325 handle.as_str(),
326 0i64,
327 updated_at_ms as i64,
328 ],
329 )
330 .map_err(process_sqlite_error)?;
331 }
332 }
333 Ok(changed)
334 })()))
335 })
336 .await
337 .map_err(process_sqlite_error)?
338 }
339
340 async fn delete_session_subscriptions(
341 &self,
342 session_id: &str,
343 ) -> Result<usize, lash_core::PluginError> {
344 let session_id = session_id.to_string();
345 self.conn
346 .write_flow(move |tx| {
347 Ok(trigger_tx_outcome((|| {
348 let mut stmt = tx
349 .prepare("SELECT subscription_id, record_json FROM trigger_subscriptions")
350 .map_err(process_sqlite_error)?;
351 let rows = stmt
352 .query_map([], |row| {
353 Ok((row.get::<_, String>(0)?, row.get::<_, String>(1)?))
354 })
355 .map_err(process_sqlite_error)?;
356 let mut subscription_ids = Vec::new();
357 for row in rows {
358 let (subscription_id, json) = row.map_err(process_sqlite_error)?;
359 let record = match Self::decode_subscription(json) {
360 Ok(record) => record,
361 Err(err) => {
362 tracing::warn!(
363 error = %err,
364 subscription_id,
365 "skipping malformed trigger subscription during session delete"
366 );
367 continue;
368 }
369 };
370 if record.registrant_session_id() == Some(session_id.as_str()) {
371 subscription_ids.push(subscription_id);
372 }
373 }
374 drop(stmt);
375 let mut deleted = 0usize;
376 for subscription_id in subscription_ids {
377 deleted = deleted.saturating_add(
378 tx.execute(
379 "DELETE FROM trigger_subscriptions WHERE subscription_id = ?1",
380 params![subscription_id.as_str()],
381 )
382 .map_err(process_sqlite_error)?,
383 );
384 }
385 Ok(deleted)
386 })()))
387 })
388 .await
389 .map_err(process_sqlite_error)?
390 }
391
392 async fn record_occurrence(
393 &self,
394 request: lash_core::TriggerOccurrenceRequest,
395 ) -> Result<lash_core::TriggerOccurrenceRecord, lash_core::PluginError> {
396 lash_core::validate_trigger_occurrence_request(&request)?;
397 let request_hash = lash_core::trigger_occurrence_request_hash(&request)?;
398 let occurrence_id = lash_core::deterministic_occurrence_id(&request)?;
399 let occurred_at_ms = self.clock.timestamp_ms();
400 self.conn
401 .write_flow(move |tx| {
402 Ok(trigger_tx_outcome((|| {
403 let existing: Option<(String, String)> = tx
404 .query_row(
405 "SELECT request_hash, record_json
406 FROM trigger_occurrences
407 WHERE idempotency_key = ?1",
408 params![request.idempotency_key.as_str()],
409 |row| Ok((row.get(0)?, row.get(1)?)),
410 )
411 .optional()
412 .map_err(process_sqlite_error)?;
413 if let Some((existing_hash, existing_json)) = existing {
414 if existing_hash != request_hash {
415 return Err(lash_core::PluginError::Session(format!(
416 "trigger occurrence idempotency conflict for `{}`",
417 request.idempotency_key
418 )));
419 }
420 return Self::decode_occurrence(existing_json);
421 }
422 let record = lash_core::TriggerOccurrenceRecord {
423 occurrence_id: occurrence_id.clone(),
424 source_type: request.source_type,
425 source_key: request.source_key,
426 payload: request.payload,
427 idempotency_key: request.idempotency_key,
428 source: request.source,
429 occurred_at_ms,
430 };
431 tx.execute(
432 "INSERT INTO trigger_occurrences (
433 occurrence_id, idempotency_key, request_hash, source_type,
434 source_key, occurred_at_ms, record_json
435 )
436 VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7)",
437 params![
438 record.occurrence_id.as_str(),
439 record.idempotency_key.as_str(),
440 request_hash.as_str(),
441 record.source_type.as_str(),
442 record.source_key.as_str(),
443 record.occurred_at_ms as i64,
444 Self::encode_json(&record)?,
445 ],
446 )
447 .map_err(process_sqlite_error)?;
448 Ok(record)
449 })()))
450 })
451 .await
452 .map_err(process_sqlite_error)?
453 }
454
455 async fn list_occurrences(
456 &self,
457 filter: lash_core::TriggerOccurrenceFilter,
458 ) -> Result<Vec<lash_core::TriggerOccurrenceRecord>, lash_core::PluginError> {
459 self.conn
460 .call(move |conn| {
461 Ok((|| {
462 let mut sql =
463 "SELECT occurrence_id, record_json FROM trigger_occurrences WHERE 1 = 1"
464 .to_string();
465 let mut values = Vec::<rusqlite::types::Value>::new();
466 if let Some(source_type) = filter.source_type.as_ref() {
467 sql.push_str(" AND source_type = ?");
468 values.push(source_type.clone().into());
469 }
470 if let Some(source_key) = filter.source_key.as_ref() {
471 sql.push_str(" AND source_key = ?");
472 values.push(source_key.clone().into());
473 }
474 if let Some(start_ms) = filter.occurred_at_start_ms {
475 sql.push_str(" AND occurred_at_ms >= ?");
476 values.push((start_ms as i64).into());
477 }
478 if let Some(end_ms) = filter.occurred_at_end_ms {
479 sql.push_str(" AND occurred_at_ms < ?");
480 values.push((end_ms as i64).into());
481 }
482 sql.push_str(" ORDER BY occurred_at_ms ASC, occurrence_id ASC");
483 let mut stmt = conn.prepare(&sql).map_err(process_sqlite_error)?;
484 let rows = stmt
485 .query_map(rusqlite::params_from_iter(values.iter()), |row| {
486 Ok((row.get::<_, String>(0)?, row.get::<_, String>(1)?))
487 })
488 .map_err(process_sqlite_error)?;
489 let mut records = Vec::new();
490 for row in rows {
491 let (occurrence_id, json) = row.map_err(process_sqlite_error)?;
492 match Self::decode_occurrence(json) {
493 Ok(record) => records.push(record),
494 Err(err) => tracing::warn!(
495 error = %err,
496 occurrence_id,
497 "skipping malformed trigger occurrence during listing"
498 ),
499 }
500 }
501 Ok(records)
502 })())
503 })
504 .await
505 .map_err(process_sqlite_error)?
506 }
507
508 async fn reserve_matching_deliveries(
509 &self,
510 occurrence_id: &str,
511 ) -> Result<Vec<lash_core::TriggerDeliveryReservation>, lash_core::PluginError> {
512 let occurrence_id = occurrence_id.to_string();
513 let created_at_ms = self.clock.timestamp_ms();
514 self.conn
515 .write_flow(move |tx| {
516 Ok(trigger_tx_outcome((|| {
517 let occurrence_json: Option<String> = tx
518 .query_row(
519 "SELECT record_json
520 FROM trigger_occurrences
521 WHERE occurrence_id = ?1",
522 params![occurrence_id.as_str()],
523 |row| row.get(0),
524 )
525 .optional()
526 .map_err(process_sqlite_error)?;
527 let Some(occurrence_json) = occurrence_json else {
528 return Err(lash_core::PluginError::Session(format!(
529 "unknown trigger occurrence `{occurrence_id}`"
530 )));
531 };
532 let occurrence = Self::decode_occurrence(occurrence_json)?;
533 let subscriptions = {
534 let mut stmt = tx
535 .prepare(
536 "SELECT subscription_id, record_json
537 FROM trigger_subscriptions
538 WHERE enabled = 1 AND source_type = ?1 AND source_key = ?2
539 ORDER BY registrant_scope_id ASC, handle ASC",
540 )
541 .map_err(process_sqlite_error)?;
542 let rows = stmt
543 .query_map(
544 params![
545 occurrence.source_type.as_str(),
546 occurrence.source_key.as_str()
547 ],
548 |row| {
549 Ok((row.get::<_, String>(0)?, row.get::<_, String>(1)?))
550 },
551 )
552 .map_err(process_sqlite_error)?;
553 let mut subscriptions = Vec::new();
554 for row in rows {
555 let (subscription_id, json) = row.map_err(process_sqlite_error)?;
556 match Self::decode_subscription(json) {
557 Ok(subscription) => subscriptions.push(subscription),
558 Err(err) => tracing::warn!(
559 error = %err,
560 subscription_id,
561 occurrence_id = %occurrence.occurrence_id,
562 "skipping malformed trigger subscription during delivery reservation"
563 ),
564 }
565 }
566 subscriptions
567 };
568 let mut reservations = Vec::new();
569 for subscription in subscriptions {
570 let process_id = lash_core::deterministic_delivery_process_id(
571 &occurrence.occurrence_id,
572 &subscription.subscription_id,
573 )?;
574 let inserted = tx
575 .execute(
576 "INSERT OR IGNORE INTO trigger_deliveries (
577 occurrence_id, subscription_id, process_id, created_at_ms
578 )
579 VALUES (?1, ?2, ?3, ?4)",
580 params![
581 occurrence.occurrence_id.as_str(),
582 subscription.subscription_id.as_str(),
583 process_id.as_str(),
584 created_at_ms as i64,
585 ],
586 )
587 .map_err(process_sqlite_error)?;
588 let stored_created_at_ms: i64 = tx
589 .query_row(
590 "SELECT created_at_ms FROM trigger_deliveries
591 WHERE occurrence_id = ?1 AND subscription_id = ?2",
592 params![
593 occurrence.occurrence_id.as_str(),
594 subscription.subscription_id.as_str()
595 ],
596 |row| row.get(0),
597 )
598 .map_err(process_sqlite_error)?;
599 reservations.push(lash_core::TriggerDeliveryReservation {
600 occurrence: occurrence.clone(),
601 subscription,
602 process_id,
603 created_at_ms: stored_created_at_ms as u64,
604 reservation_status: if inserted == 0 {
605 lash_core::TriggerDeliveryReservationStatus::AlreadyReserved
606 } else {
607 lash_core::TriggerDeliveryReservationStatus::Reserved
608 },
609 });
610 }
611 Ok(reservations)
612 })()))
613 })
614 .await
615 .map_err(process_sqlite_error)?
616 }
617
618 async fn list_deliveries_by_occurrence_id(
619 &self,
620 occurrence_id: &str,
621 ) -> Result<Vec<lash_core::TriggerDeliveryReservation>, lash_core::PluginError> {
622 self.list_deliveries_where("d.occurrence_id = ?1", occurrence_id.to_string())
623 .await
624 }
625
626 async fn list_deliveries_by_subscription_id(
627 &self,
628 subscription_id: &str,
629 ) -> Result<Vec<lash_core::TriggerDeliveryReservation>, lash_core::PluginError> {
630 self.list_deliveries_where("d.subscription_id = ?1", subscription_id.to_string())
631 .await
632 }
633
634 async fn list_deliveries_by_process_id(
635 &self,
636 process_id: &str,
637 ) -> Result<Vec<lash_core::TriggerDeliveryReservation>, lash_core::PluginError> {
638 self.list_deliveries_where("d.process_id = ?1", process_id.to_string())
639 .await
640 }
641}