1use super::*;
8
9pub struct SqliteTriggerStore {
10 conn: SqliteConnection,
11}
12
13impl SqliteTriggerStore {
14 pub async fn open(path: &Path) -> tokio_rusqlite::Result<Self> {
15 let conn = SqliteConnection::open(path).await?;
16 ensure_trigger_schema(&conn).await?;
17 apply_pragmas(&conn, StoreBacking::File).await?;
18 Ok(Self { conn })
19 }
20
21 pub async fn memory() -> tokio_rusqlite::Result<Self> {
22 let conn = SqliteConnection::open_in_memory().await?;
23 ensure_trigger_schema(&conn).await?;
24 apply_pragmas(&conn, StoreBacking::Memory).await?;
25 Ok(Self { conn })
26 }
27
28 fn encode_json<T: serde::Serialize>(value: &T) -> Result<String, lash_core::PluginError> {
29 serde_json::to_string(value).map_err(|err| {
30 lash_core::PluginError::Session(format!("failed to encode trigger row: {err}"))
31 })
32 }
33
34 fn decode_subscription(
35 json: String,
36 ) -> Result<lash_core::TriggerSubscriptionRecord, lash_core::PluginError> {
37 serde_json::from_str(&json).map_err(|err| {
38 lash_core::PluginError::Session(format!(
39 "failed to decode trigger subscription row: {err}"
40 ))
41 })
42 }
43
44 fn decode_occurrence(
45 json: String,
46 ) -> Result<lash_core::TriggerOccurrenceRecord, lash_core::PluginError> {
47 serde_json::from_str(&json).map_err(|err| {
48 lash_core::PluginError::Session(format!(
49 "failed to decode trigger occurrence row: {err}"
50 ))
51 })
52 }
53
54 fn decode_delivery(
55 occurrence_json: String,
56 subscription_json: String,
57 process_id: String,
58 created_at_ms: i64,
59 reservation_status: lash_core::TriggerDeliveryReservationStatus,
60 ) -> Result<lash_core::TriggerDeliveryReservation, lash_core::PluginError> {
61 Ok(lash_core::TriggerDeliveryReservation {
62 occurrence: Self::decode_occurrence(occurrence_json)?,
63 subscription: Self::decode_subscription(subscription_json)?,
64 process_id,
65 created_at_ms: created_at_ms as u64,
66 reservation_status,
67 })
68 }
69
70 async fn list_deliveries_where(
71 &self,
72 where_clause: &'static str,
73 value: String,
74 ) -> Result<Vec<lash_core::TriggerDeliveryReservation>, lash_core::PluginError> {
75 self.conn
76 .call(move |conn| {
77 Ok((|| {
78 let sql = format!(
79 "SELECT d.process_id, d.created_at_ms, o.record_json, s.record_json
80 FROM trigger_deliveries d
81 JOIN trigger_occurrences o ON o.occurrence_id = d.occurrence_id
82 JOIN trigger_subscriptions s ON s.subscription_id = d.subscription_id
83 WHERE {where_clause}
84 ORDER BY d.created_at_ms ASC, d.occurrence_id ASC, d.subscription_id ASC"
85 );
86 let mut stmt = conn.prepare(&sql).map_err(process_sqlite_error)?;
87 let rows = stmt
88 .query_map(params![value.as_str()], |row| {
89 Ok((
90 row.get::<_, String>(0)?,
91 row.get::<_, i64>(1)?,
92 row.get::<_, String>(2)?,
93 row.get::<_, String>(3)?,
94 ))
95 })
96 .map_err(process_sqlite_error)?;
97 let mut deliveries = Vec::new();
98 for row in rows {
99 let (process_id, created_at_ms, occurrence_json, subscription_json) =
100 row.map_err(process_sqlite_error)?;
101 deliveries.push(Self::decode_delivery(
102 occurrence_json,
103 subscription_json,
104 process_id,
105 created_at_ms,
106 lash_core::TriggerDeliveryReservationStatus::AlreadyReserved,
107 )?);
108 }
109 Ok(deliveries)
110 })())
111 })
112 .await
113 .map_err(process_sqlite_error)?
114 }
115}
116
117fn trigger_tx_outcome<T>(
118 result: Result<T, lash_core::PluginError>,
119) -> TxOutcome<Result<T, lash_core::PluginError>> {
120 match result {
121 Ok(value) => TxOutcome::Commit(Ok(value)),
122 Err(err) => TxOutcome::Rollback(Err(err)),
123 }
124}
125
126#[async_trait::async_trait]
127impl lash_core::TriggerStore for SqliteTriggerStore {
128 fn durability_tier(&self) -> DurabilityTier {
129 DurabilityTier::Durable
130 }
131
132 async fn register_subscription(
133 &self,
134 draft: lash_core::TriggerSubscriptionDraft,
135 ) -> Result<lash_core::TriggerSubscriptionRecord, lash_core::PluginError> {
136 draft.validate()?;
137 self.conn
138 .write_flow(move |tx| {
139 Ok(trigger_tx_outcome((|| {
140 tx.execute("INSERT INTO trigger_subscription_seq DEFAULT VALUES", [])
141 .map_err(process_sqlite_error)?;
142 let seq = tx.last_insert_rowid();
143 let handle = format!("trigger:{seq}");
144 let subscription_id = format!("subscription:{seq}");
145 let now = current_epoch_ms();
146 let record = lash_core::TriggerSubscriptionRecord {
147 subscription_id: subscription_id.clone(),
148 registrant: draft.registrant,
149 env_ref: draft.env_ref,
150 wake_target: draft.wake_target,
151 handle,
152 name: draft.name,
153 source_type: draft.source_type,
154 source_key: draft.source_key,
155 source: draft.source,
156 payload_schema: draft.payload_schema,
157 target: draft.target,
158 target_identity: draft.target_identity,
159 event_types: draft.event_types,
160 input_template: draft.input_template,
161 target_label: draft.target_label,
162 enabled: true,
163 created_at_ms: now,
164 updated_at_ms: now,
165 };
166 tx.execute(
167 "INSERT INTO trigger_subscriptions (
168 subscription_id, registrant_scope_id, handle, source_type, source_key,
169 enabled, created_at_ms, updated_at_ms, record_json
170 )
171 VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9)",
172 params![
173 record.subscription_id.as_str(),
174 record.registrant_scope_id().as_str(),
175 record.handle.as_str(),
176 record.source_type.as_str(),
177 record.source_key.as_str(),
178 i64::from(record.enabled),
179 record.created_at_ms as i64,
180 record.updated_at_ms as i64,
181 Self::encode_json(&record)?,
182 ],
183 )
184 .map_err(process_sqlite_error)?;
185 Ok(record)
186 })()))
187 })
188 .await
189 .map_err(process_sqlite_error)?
190 }
191
192 async fn list_subscriptions(
193 &self,
194 filter: lash_core::TriggerSubscriptionFilter,
195 ) -> Result<Vec<lash_core::TriggerSubscriptionRecord>, lash_core::PluginError> {
196 self.conn
197 .call(move |conn| {
198 Ok((|| {
199 let mut sql =
200 "SELECT subscription_id, record_json FROM trigger_subscriptions WHERE 1 = 1"
201 .to_string();
202 let mut values = Vec::<rusqlite::types::Value>::new();
203 if let Some(registrant_scope_id) = filter.effective_registrant_scope_id() {
204 sql.push_str(" AND registrant_scope_id = ?");
205 values.push(registrant_scope_id.into());
206 }
207 if let Some(handle) = filter.handle.as_ref() {
208 sql.push_str(" AND handle = ?");
209 values.push(handle.clone().into());
210 }
211 if let Some(source_type) = filter.source_type.as_ref() {
212 sql.push_str(" AND source_type = ?");
213 values.push(source_type.clone().into());
214 }
215 if let Some(source_key) = filter.source_key.as_ref() {
216 sql.push_str(" AND source_key = ?");
217 values.push(source_key.clone().into());
218 }
219 if let Some(enabled) = filter.enabled {
220 sql.push_str(" AND enabled = ?");
221 values.push(i64::from(enabled).into());
222 }
223 sql.push_str(" ORDER BY registrant_scope_id ASC, handle ASC");
224 let mut stmt = conn.prepare(&sql).map_err(process_sqlite_error)?;
225 let rows = stmt
226 .query_map(rusqlite::params_from_iter(values.iter()), |row| {
227 Ok((row.get::<_, String>(0)?, row.get::<_, String>(1)?))
228 })
229 .map_err(process_sqlite_error)?;
230 let mut records = Vec::new();
231 for row in rows {
232 let (subscription_id, json) = row.map_err(process_sqlite_error)?;
233 let record = match Self::decode_subscription(json) {
234 Ok(record) => record,
235 Err(err) => {
236 tracing::warn!(
237 error = %err,
238 subscription_id,
239 "skipping malformed trigger subscription during listing"
240 );
241 continue;
242 }
243 };
244 if filter.matches(&record) {
245 records.push(record);
246 }
247 }
248 Ok(records)
249 })())
250 })
251 .await
252 .map_err(process_sqlite_error)?
253 }
254
255 async fn cancel_subscription(
256 &self,
257 registrant_scope_id: &str,
258 handle: &str,
259 ) -> Result<bool, lash_core::PluginError> {
260 let registrant_scope_id = registrant_scope_id.to_string();
261 let handle = handle.to_string();
262 self.conn
263 .write_flow(move |tx| {
264 Ok(trigger_tx_outcome((|| {
265 let selected: Option<(String, i64, String)> = tx
266 .query_row(
267 "SELECT subscription_id, enabled, record_json
268 FROM trigger_subscriptions
269 WHERE registrant_scope_id = ?1 AND handle = ?2",
270 params![registrant_scope_id.as_str(), handle.as_str()],
271 |row| Ok((row.get(0)?, row.get(1)?, row.get(2)?)),
272 )
273 .optional()
274 .map_err(process_sqlite_error)?;
275 let Some((subscription_id, enabled, json)) = selected else {
276 return Ok(false);
277 };
278 let changed = enabled != 0;
279 let updated_at_ms = current_epoch_ms();
280 match Self::decode_subscription(json) {
281 Ok(mut record) => {
282 record.enabled = false;
283 record.updated_at_ms = updated_at_ms;
284 tx.execute(
285 "UPDATE trigger_subscriptions
286 SET enabled = ?3, updated_at_ms = ?4, record_json = ?5
287 WHERE subscription_id = ?1 AND handle = ?2",
288 params![
289 subscription_id.as_str(),
290 handle.as_str(),
291 i64::from(record.enabled),
292 record.updated_at_ms as i64,
293 Self::encode_json(&record)?,
294 ],
295 )
296 .map_err(process_sqlite_error)?;
297 }
298 Err(err) => {
299 tracing::warn!(
300 error = %err,
301 subscription_id,
302 handle,
303 "disabling malformed trigger subscription without rewriting record JSON"
304 );
305 tx.execute(
306 "UPDATE trigger_subscriptions
307 SET enabled = ?3, updated_at_ms = ?4
308 WHERE subscription_id = ?1 AND handle = ?2",
309 params![
310 subscription_id.as_str(),
311 handle.as_str(),
312 0i64,
313 updated_at_ms as i64,
314 ],
315 )
316 .map_err(process_sqlite_error)?;
317 }
318 }
319 Ok(changed)
320 })()))
321 })
322 .await
323 .map_err(process_sqlite_error)?
324 }
325
326 async fn delete_session_subscriptions(
327 &self,
328 session_id: &str,
329 ) -> Result<usize, lash_core::PluginError> {
330 let session_id = session_id.to_string();
331 self.conn
332 .write_flow(move |tx| {
333 Ok(trigger_tx_outcome((|| {
334 let mut stmt = tx
335 .prepare("SELECT subscription_id, record_json FROM trigger_subscriptions")
336 .map_err(process_sqlite_error)?;
337 let rows = stmt
338 .query_map([], |row| {
339 Ok((row.get::<_, String>(0)?, row.get::<_, String>(1)?))
340 })
341 .map_err(process_sqlite_error)?;
342 let mut subscription_ids = Vec::new();
343 for row in rows {
344 let (subscription_id, json) = row.map_err(process_sqlite_error)?;
345 let record = match Self::decode_subscription(json) {
346 Ok(record) => record,
347 Err(err) => {
348 tracing::warn!(
349 error = %err,
350 subscription_id,
351 "skipping malformed trigger subscription during session delete"
352 );
353 continue;
354 }
355 };
356 if record.registrant_session_id() == Some(session_id.as_str()) {
357 subscription_ids.push(subscription_id);
358 }
359 }
360 drop(stmt);
361 let mut deleted = 0usize;
362 for subscription_id in subscription_ids {
363 deleted = deleted.saturating_add(
364 tx.execute(
365 "DELETE FROM trigger_subscriptions WHERE subscription_id = ?1",
366 params![subscription_id.as_str()],
367 )
368 .map_err(process_sqlite_error)?,
369 );
370 }
371 Ok(deleted)
372 })()))
373 })
374 .await
375 .map_err(process_sqlite_error)?
376 }
377
378 async fn record_occurrence(
379 &self,
380 request: lash_core::TriggerOccurrenceRequest,
381 ) -> Result<lash_core::TriggerOccurrenceRecord, lash_core::PluginError> {
382 lash_core::validate_trigger_occurrence_request(&request)?;
383 let request_hash = lash_core::trigger_occurrence_request_hash(&request)?;
384 let occurrence_id = lash_core::deterministic_occurrence_id(&request)?;
385 self.conn
386 .write_flow(move |tx| {
387 Ok(trigger_tx_outcome((|| {
388 let existing: Option<(String, String)> = tx
389 .query_row(
390 "SELECT request_hash, record_json
391 FROM trigger_occurrences
392 WHERE idempotency_key = ?1",
393 params![request.idempotency_key.as_str()],
394 |row| Ok((row.get(0)?, row.get(1)?)),
395 )
396 .optional()
397 .map_err(process_sqlite_error)?;
398 if let Some((existing_hash, existing_json)) = existing {
399 if existing_hash != request_hash {
400 return Err(lash_core::PluginError::Session(format!(
401 "trigger occurrence idempotency conflict for `{}`",
402 request.idempotency_key
403 )));
404 }
405 return Self::decode_occurrence(existing_json);
406 }
407 let record = lash_core::TriggerOccurrenceRecord {
408 occurrence_id: occurrence_id.clone(),
409 source_type: request.source_type,
410 source_key: request.source_key,
411 payload: request.payload,
412 idempotency_key: request.idempotency_key,
413 source: request.source,
414 occurred_at_ms: current_epoch_ms(),
415 };
416 tx.execute(
417 "INSERT INTO trigger_occurrences (
418 occurrence_id, idempotency_key, request_hash, source_type,
419 source_key, occurred_at_ms, record_json
420 )
421 VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7)",
422 params![
423 record.occurrence_id.as_str(),
424 record.idempotency_key.as_str(),
425 request_hash.as_str(),
426 record.source_type.as_str(),
427 record.source_key.as_str(),
428 record.occurred_at_ms as i64,
429 Self::encode_json(&record)?,
430 ],
431 )
432 .map_err(process_sqlite_error)?;
433 Ok(record)
434 })()))
435 })
436 .await
437 .map_err(process_sqlite_error)?
438 }
439
440 async fn list_occurrences(
441 &self,
442 filter: lash_core::TriggerOccurrenceFilter,
443 ) -> Result<Vec<lash_core::TriggerOccurrenceRecord>, lash_core::PluginError> {
444 self.conn
445 .call(move |conn| {
446 Ok((|| {
447 let mut sql =
448 "SELECT occurrence_id, record_json FROM trigger_occurrences WHERE 1 = 1"
449 .to_string();
450 let mut values = Vec::<rusqlite::types::Value>::new();
451 if let Some(source_type) = filter.source_type.as_ref() {
452 sql.push_str(" AND source_type = ?");
453 values.push(source_type.clone().into());
454 }
455 if let Some(source_key) = filter.source_key.as_ref() {
456 sql.push_str(" AND source_key = ?");
457 values.push(source_key.clone().into());
458 }
459 if let Some(start_ms) = filter.occurred_at_start_ms {
460 sql.push_str(" AND occurred_at_ms >= ?");
461 values.push((start_ms as i64).into());
462 }
463 if let Some(end_ms) = filter.occurred_at_end_ms {
464 sql.push_str(" AND occurred_at_ms < ?");
465 values.push((end_ms as i64).into());
466 }
467 sql.push_str(" ORDER BY occurred_at_ms ASC, occurrence_id ASC");
468 let mut stmt = conn.prepare(&sql).map_err(process_sqlite_error)?;
469 let rows = stmt
470 .query_map(rusqlite::params_from_iter(values.iter()), |row| {
471 Ok((row.get::<_, String>(0)?, row.get::<_, String>(1)?))
472 })
473 .map_err(process_sqlite_error)?;
474 let mut records = Vec::new();
475 for row in rows {
476 let (occurrence_id, json) = row.map_err(process_sqlite_error)?;
477 match Self::decode_occurrence(json) {
478 Ok(record) => records.push(record),
479 Err(err) => tracing::warn!(
480 error = %err,
481 occurrence_id,
482 "skipping malformed trigger occurrence during listing"
483 ),
484 }
485 }
486 Ok(records)
487 })())
488 })
489 .await
490 .map_err(process_sqlite_error)?
491 }
492
493 async fn reserve_matching_deliveries(
494 &self,
495 occurrence_id: &str,
496 ) -> Result<Vec<lash_core::TriggerDeliveryReservation>, lash_core::PluginError> {
497 let occurrence_id = occurrence_id.to_string();
498 self.conn
499 .write_flow(move |tx| {
500 Ok(trigger_tx_outcome((|| {
501 let occurrence_json: Option<String> = tx
502 .query_row(
503 "SELECT record_json
504 FROM trigger_occurrences
505 WHERE occurrence_id = ?1",
506 params![occurrence_id.as_str()],
507 |row| row.get(0),
508 )
509 .optional()
510 .map_err(process_sqlite_error)?;
511 let Some(occurrence_json) = occurrence_json else {
512 return Err(lash_core::PluginError::Session(format!(
513 "unknown trigger occurrence `{occurrence_id}`"
514 )));
515 };
516 let occurrence = Self::decode_occurrence(occurrence_json)?;
517 let subscriptions = {
518 let mut stmt = tx
519 .prepare(
520 "SELECT subscription_id, record_json
521 FROM trigger_subscriptions
522 WHERE enabled = 1 AND source_type = ?1 AND source_key = ?2
523 ORDER BY registrant_scope_id ASC, handle ASC",
524 )
525 .map_err(process_sqlite_error)?;
526 let rows = stmt
527 .query_map(
528 params![
529 occurrence.source_type.as_str(),
530 occurrence.source_key.as_str()
531 ],
532 |row| {
533 Ok((row.get::<_, String>(0)?, row.get::<_, String>(1)?))
534 },
535 )
536 .map_err(process_sqlite_error)?;
537 let mut subscriptions = Vec::new();
538 for row in rows {
539 let (subscription_id, json) = row.map_err(process_sqlite_error)?;
540 match Self::decode_subscription(json) {
541 Ok(subscription) => subscriptions.push(subscription),
542 Err(err) => tracing::warn!(
543 error = %err,
544 subscription_id,
545 occurrence_id = %occurrence.occurrence_id,
546 "skipping malformed trigger subscription during delivery reservation"
547 ),
548 }
549 }
550 subscriptions
551 };
552 let mut reservations = Vec::new();
553 for subscription in subscriptions {
554 let process_id = lash_core::deterministic_delivery_process_id(
555 &occurrence.occurrence_id,
556 &subscription.subscription_id,
557 )?;
558 let created_at_ms = current_epoch_ms();
559 let inserted = tx
560 .execute(
561 "INSERT OR IGNORE INTO trigger_deliveries (
562 occurrence_id, subscription_id, process_id, created_at_ms
563 )
564 VALUES (?1, ?2, ?3, ?4)",
565 params![
566 occurrence.occurrence_id.as_str(),
567 subscription.subscription_id.as_str(),
568 process_id.as_str(),
569 created_at_ms as i64,
570 ],
571 )
572 .map_err(process_sqlite_error)?;
573 let stored_created_at_ms: i64 = tx
574 .query_row(
575 "SELECT created_at_ms FROM trigger_deliveries
576 WHERE occurrence_id = ?1 AND subscription_id = ?2",
577 params![
578 occurrence.occurrence_id.as_str(),
579 subscription.subscription_id.as_str()
580 ],
581 |row| row.get(0),
582 )
583 .map_err(process_sqlite_error)?;
584 reservations.push(lash_core::TriggerDeliveryReservation {
585 occurrence: occurrence.clone(),
586 subscription,
587 process_id,
588 created_at_ms: stored_created_at_ms as u64,
589 reservation_status: if inserted == 0 {
590 lash_core::TriggerDeliveryReservationStatus::AlreadyReserved
591 } else {
592 lash_core::TriggerDeliveryReservationStatus::Reserved
593 },
594 });
595 }
596 Ok(reservations)
597 })()))
598 })
599 .await
600 .map_err(process_sqlite_error)?
601 }
602
603 async fn list_deliveries_by_occurrence_id(
604 &self,
605 occurrence_id: &str,
606 ) -> Result<Vec<lash_core::TriggerDeliveryReservation>, lash_core::PluginError> {
607 self.list_deliveries_where("d.occurrence_id = ?1", occurrence_id.to_string())
608 .await
609 }
610
611 async fn list_deliveries_by_subscription_id(
612 &self,
613 subscription_id: &str,
614 ) -> Result<Vec<lash_core::TriggerDeliveryReservation>, lash_core::PluginError> {
615 self.list_deliveries_where("d.subscription_id = ?1", subscription_id.to_string())
616 .await
617 }
618
619 async fn list_deliveries_by_process_id(
620 &self,
621 process_id: &str,
622 ) -> Result<Vec<lash_core::TriggerDeliveryReservation>, lash_core::PluginError> {
623 self.list_deliveries_where("d.process_id = ?1", process_id.to_string())
624 .await
625 }
626}