aion_store_libsql/
store.rs

1//! `LibSqlStore` struct and `EventStore` implementation wiring.
2
3use std::{path::PathBuf, sync::Arc};
4
5use aion_store::{
6    ClaimScope, Event, MintOutcome, NamespaceOrigin, NamespacePlacement, NamespaceRecord,
7    NamespaceStore, OutboxRow, OutboxStore, PackageRecord, PackageRouteRecord, PackageStore,
8    ReadableEventStore, RunSummary, StoreError, TimerEntry, TimerId, WorkflowFilter, WorkflowId,
9    WorkflowSummary, WritableEventStore, WriteToken,
10};
11use async_trait::async_trait;
12use chrono::{DateTime, Utc};
13use tokio::sync::Mutex;
14
15use crate::config::{LibSqlConfig, LibSqlMode};
16use crate::outbox::OutboxRowState;
17
18/// Durable `EventStore` backed by a shared libSQL connection.
19#[derive(Clone)]
20pub struct LibSqlStore {
21    conn: libsql::Connection,
22    db: Arc<libsql::Database>,
23    transaction_lock: Arc<Mutex<()>>,
24}
25
26impl LibSqlStore {
27    /// Open a store from operator-provided libSQL configuration.
28    ///
29    /// # Errors
30    ///
31    /// Returns `StoreError::Backend` when the connection cannot be opened or when the idempotent
32    /// schema DDL cannot be applied.
33    pub async fn connect(config: LibSqlConfig) -> Result<Self, StoreError> {
34        let opened = crate::connection::open_connection(&config).await?;
35        let conn = opened.connection;
36        crate::schema::ensure_schema(&conn).await?;
37
38        Ok(Self {
39            conn,
40            db: Arc::new(opened.database),
41            transaction_lock: Arc::new(Mutex::new(())),
42        })
43    }
44
45    /// Open an embedded local-file store at `path`.
46    ///
47    /// Operator tunables remain unset; this convenience constructor only selects embedded mode.
48    ///
49    /// # Errors
50    ///
51    /// Returns `StoreError::Backend` when the connection cannot be opened or when the idempotent
52    /// schema DDL cannot be applied.
53    pub async fn open(path: impl Into<PathBuf>) -> Result<Self, StoreError> {
54        Self::connect(LibSqlConfig {
55            mode: LibSqlMode::Embedded { path: path.into() },
56            journal_mode: None,
57            synchronous: None,
58            sync_interval_seconds: None,
59        })
60        .await
61    }
62
63    /// Validate stored event blobs against the current Aion event schema.
64    ///
65    /// # Errors
66    ///
67    /// Returns `StoreError::Serialization` when any stored event cannot be decoded by the current
68    /// event schema, or `StoreError::Backend` for libSQL scan failures.
69    pub async fn validate_event_compatibility(&self) -> Result<(), StoreError> {
70        crate::read::validate_all_events(self.connection()).await
71    }
72
73    /// Trigger and await a libSQL replica synchronization cycle.
74    ///
75    /// # Errors
76    ///
77    /// Returns `StoreError::Backend` when the current libSQL database mode does not support sync or
78    /// when the replica sync operation fails.
79    pub async fn sync(&self) -> Result<(), StoreError> {
80        self.db
81            .sync()
82            .await
83            .map(|_| ())
84            .map_err(|error| crate::error::libsql_error(&error))
85    }
86
87    /// Atomically append `events` and, when `outbox_rows` is `Some`, the outbox rows in the same
88    /// `IMMEDIATE` transaction, under the expected-head sequence guard.
89    ///
90    /// This is the durable fan-out write: the scheduling events and their outbox rows commit
91    /// together or not at all. Passing `None` is equivalent to
92    /// [`WritableEventStore::append`](aion_store::WritableEventStore::append).
93    ///
94    /// # Errors
95    ///
96    /// Returns `StoreError::SequenceConflict` when the stored head differs from `expected_seq`,
97    /// `StoreError::Serialization` when an event or outbox payload cannot be serialized, and
98    /// `StoreError::Backend` for libSQL boundary failures.
99    pub async fn append_with_outbox(
100        &self,
101        _token: WriteToken,
102        workflow_id: &WorkflowId,
103        events: &[Event],
104        expected_seq: u64,
105        outbox_rows: Option<&[OutboxRow]>,
106    ) -> Result<(), StoreError> {
107        let _guard = self.transaction_lock.lock().await;
108        crate::append::append_with_outbox(
109            self.connection(),
110            workflow_id,
111            events,
112            expected_seq,
113            outbox_rows,
114        )
115        .await
116    }
117
118    /// Read the persisted lifecycle state of one outbox row by its dispatch key.
119    ///
120    /// Returns the `(status, attempt, visible_after)` triple, or `None` when no
121    /// row carries `dispatch_key` (the dedup guard may have ignored it). This is
122    /// an out-of-band inspection helper — the [`OutboxStore`] dispatch contract
123    /// itself keys terminal transitions off `dispatch_key` and never needs to
124    /// read a row back — used by the dispatcher's tests and by operators auditing
125    /// dead-lettered (`failed`) rows.
126    ///
127    /// # Errors
128    ///
129    /// Returns `StoreError::Backend` for libSQL boundary failures and
130    /// `StoreError::Serialization` when the stored status token or timestamp
131    /// cannot be decoded.
132    pub async fn outbox_row_state(
133        &self,
134        dispatch_key: &str,
135    ) -> Result<Option<OutboxRowState>, StoreError> {
136        crate::outbox::outbox_row_state(self.connection(), dispatch_key).await
137    }
138
139    /// Borrow the shared libSQL connection used by append, read, and timer modules.
140    pub(crate) fn connection(&self) -> &libsql::Connection {
141        &self.conn
142    }
143}
144
145#[async_trait]
146impl PackageStore for LibSqlStore {
147    async fn put_package(&self, record: PackageRecord) -> Result<(), StoreError> {
148        crate::package::put_package(self.connection(), record).await
149    }
150
151    async fn put_package_with_routes(
152        &self,
153        record: PackageRecord,
154        route_workflow_types: &[String],
155    ) -> Result<(), StoreError> {
156        crate::package::put_package_with_routes(self.connection(), record, route_workflow_types)
157            .await
158    }
159
160    async fn list_packages(&self) -> Result<Vec<PackageRecord>, StoreError> {
161        crate::package::list_packages(self.connection()).await
162    }
163
164    async fn delete_package(
165        &self,
166        workflow_type: &str,
167        content_hash: &str,
168    ) -> Result<(), StoreError> {
169        crate::package::delete_package(self.connection(), workflow_type, content_hash).await
170    }
171
172    async fn put_package_route(
173        &self,
174        workflow_type: &str,
175        content_hash: &str,
176    ) -> Result<(), StoreError> {
177        crate::package::put_package_route(self.connection(), workflow_type, content_hash).await
178    }
179
180    async fn list_package_routes(&self) -> Result<Vec<PackageRouteRecord>, StoreError> {
181        crate::package::list_package_routes(self.connection()).await
182    }
183}
184
185#[async_trait]
186impl NamespaceStore for LibSqlStore {
187    async fn register_namespace(
188        &self,
189        name: &str,
190        origin: NamespaceOrigin,
191    ) -> Result<MintOutcome, StoreError> {
192        crate::namespace::register_namespace(self.connection(), name, origin).await
193    }
194
195    async fn put_namespace(&self, record: NamespaceRecord) -> Result<MintOutcome, StoreError> {
196        crate::namespace::put_namespace(self.connection(), record).await
197    }
198
199    async fn list_namespaces(&self) -> Result<Vec<NamespaceRecord>, StoreError> {
200        crate::namespace::list_namespaces(self.connection()).await
201    }
202
203    async fn get_namespace(&self, name: &str) -> Result<Option<NamespaceRecord>, StoreError> {
204        crate::namespace::get_namespace(self.connection(), name).await
205    }
206
207    async fn set_namespace_placement(
208        &self,
209        name: &str,
210        placement: NamespacePlacement,
211    ) -> Result<Option<()>, StoreError> {
212        crate::namespace::set_namespace_placement(self.connection(), name, placement).await
213    }
214
215    async fn deprecate_namespace(&self, name: &str) -> Result<(), StoreError> {
216        crate::namespace::deprecate_namespace(self.connection(), name).await
217    }
218}
219
220#[async_trait]
221impl WritableEventStore for LibSqlStore {
222    async fn append(
223        &self,
224        _token: WriteToken,
225        workflow_id: &WorkflowId,
226        events: &[Event],
227        expected_seq: u64,
228    ) -> Result<(), StoreError> {
229        let _guard = self.transaction_lock.lock().await;
230        crate::append::append(self.connection(), workflow_id, events, expected_seq).await
231    }
232
233    async fn append_with_outbox(
234        &self,
235        _token: WriteToken,
236        workflow_id: &WorkflowId,
237        events: &[Event],
238        expected_seq: u64,
239        outbox_rows: &[OutboxRow],
240    ) -> Result<(), StoreError> {
241        let _guard = self.transaction_lock.lock().await;
242        crate::append::append_with_outbox(
243            self.connection(),
244            workflow_id,
245            events,
246            expected_seq,
247            Some(outbox_rows),
248        )
249        .await
250    }
251
252    async fn rearm_outbox_pending(&self, rows: &[OutboxRow]) -> Result<(), StoreError> {
253        let _guard = self.transaction_lock.lock().await;
254        crate::outbox::rearm_outbox_pending(self.connection(), rows).await
255    }
256
257    async fn settle_outbox_row_cancelled(&self, dispatch_key: &str) -> Result<(), StoreError> {
258        crate::outbox::settle_outbox_row_cancelled(self.connection(), dispatch_key).await
259    }
260
261    async fn settle_workflow_outbox_rows_cancelled(
262        &self,
263        workflow_id: &WorkflowId,
264    ) -> Result<Vec<String>, StoreError> {
265        let _guard = self.transaction_lock.lock().await;
266        crate::outbox::settle_workflow_outbox_rows_cancelled(self.connection(), workflow_id).await
267    }
268}
269
270#[async_trait]
271impl OutboxStore for LibSqlStore {
272    async fn append_outbox_batch(&self, rows: &[OutboxRow]) -> Result<(), StoreError> {
273        let _guard = self.transaction_lock.lock().await;
274        crate::outbox::append_outbox_batch(self.connection(), rows).await
275    }
276
277    async fn claim_outbox_rows(&self, limit: u32) -> Result<Vec<OutboxRow>, StoreError> {
278        let _guard = self.transaction_lock.lock().await;
279        crate::outbox::claim_outbox_rows(self.connection(), limit).await
280    }
281
282    async fn claim_outbox_rows_excluding(
283        &self,
284        limit: u32,
285        held: &std::collections::HashSet<WorkflowId>,
286    ) -> Result<Vec<OutboxRow>, StoreError> {
287        let _guard = self.transaction_lock.lock().await;
288        crate::outbox::claim_outbox_rows_excluding(self.connection(), limit, held).await
289    }
290
291    async fn claim_outbox_rows_scoped(
292        &self,
293        scope: &ClaimScope,
294        limit: u32,
295    ) -> Result<Vec<OutboxRow>, StoreError> {
296        let _guard = self.transaction_lock.lock().await;
297        crate::outbox::claim_outbox_rows_scoped(self.connection(), scope, limit).await
298    }
299
300    async fn claim_outbox_rows_scoped_excluding(
301        &self,
302        scope: &ClaimScope,
303        limit: u32,
304        held: &std::collections::HashSet<WorkflowId>,
305    ) -> Result<Vec<OutboxRow>, StoreError> {
306        let _guard = self.transaction_lock.lock().await;
307        crate::outbox::claim_outbox_rows_scoped_excluding(self.connection(), scope, limit, held)
308            .await
309    }
310
311    async fn rearm_stale_claimed_outbox_rows(
312        &self,
313        older_than: DateTime<Utc>,
314        visible_after: DateTime<Utc>,
315        limit: u32,
316    ) -> Result<Vec<OutboxRow>, StoreError> {
317        let _guard = self.transaction_lock.lock().await;
318        crate::outbox::rearm_stale_claimed_outbox_rows(
319            self.connection(),
320            older_than,
321            visible_after,
322            limit,
323        )
324        .await
325    }
326
327    async fn list_stale_claimed_outbox_rows(
328        &self,
329        older_than: DateTime<Utc>,
330        limit: u32,
331    ) -> Result<Vec<OutboxRow>, StoreError> {
332        crate::outbox::list_stale_claimed_outbox_rows(self.connection(), older_than, limit).await
333    }
334
335    async fn list_unsettled_outbox_workflow_ids(&self) -> Result<Vec<WorkflowId>, StoreError> {
336        crate::outbox::list_unsettled_outbox_workflow_ids(self.connection()).await
337    }
338
339    async fn cancel_outbox_rows_for_workflow(
340        &self,
341        workflow_id: &WorkflowId,
342    ) -> Result<Vec<String>, StoreError> {
343        let _guard = self.transaction_lock.lock().await;
344        crate::outbox::settle_workflow_outbox_rows_cancelled(self.connection(), workflow_id).await
345    }
346
347    async fn complete_outbox_row(&self, dispatch_key: &str) -> Result<(), StoreError> {
348        crate::outbox::complete_outbox_row(self.connection(), dispatch_key).await
349    }
350
351    async fn retry_outbox_row(
352        &self,
353        dispatch_key: &str,
354        next_attempt: u32,
355        visible_after: DateTime<Utc>,
356    ) -> Result<(), StoreError> {
357        crate::outbox::retry_outbox_row(
358            self.connection(),
359            dispatch_key,
360            next_attempt,
361            visible_after,
362        )
363        .await
364    }
365
366    async fn fail_outbox_row(&self, dispatch_key: &str) -> Result<(), StoreError> {
367        crate::outbox::fail_outbox_row(self.connection(), dispatch_key).await
368    }
369
370    async fn count_inflight_outbox_rows(&self, namespace: &str) -> Result<u64, StoreError> {
371        crate::outbox::count_inflight_outbox_rows(self.connection(), namespace).await
372    }
373
374    async fn count_claimed_outbox_rows(&self, namespace: &str) -> Result<u64, StoreError> {
375        crate::outbox::count_claimed_outbox_rows(self.connection(), namespace).await
376    }
377
378    async fn count_claimed_outbox_rows_by_namespace(
379        &self,
380        namespaces: &[&str],
381    ) -> Result<std::collections::BTreeMap<String, u64>, StoreError> {
382        crate::outbox::count_claimed_outbox_rows_by_namespace(self.connection(), namespaces).await
383    }
384
385    async fn pending_outbox_routes(&self) -> Result<Vec<ClaimScope>, StoreError> {
386        crate::outbox::pending_outbox_routes(self.connection()).await
387    }
388}
389
390#[async_trait]
391impl ReadableEventStore for LibSqlStore {
392    async fn read_history(&self, workflow_id: &WorkflowId) -> Result<Vec<Event>, StoreError> {
393        crate::read::read_history(self.connection(), workflow_id).await
394    }
395
396    async fn read_history_from(
397        &self,
398        workflow_id: &WorkflowId,
399        from_seq: u64,
400    ) -> Result<Vec<Event>, StoreError> {
401        crate::read::read_history_from(self.connection(), workflow_id, from_seq).await
402    }
403
404    async fn read_run_chain(
405        &self,
406        workflow_id: &WorkflowId,
407    ) -> Result<Vec<RunSummary>, StoreError> {
408        crate::read::read_run_chain(self.connection(), workflow_id).await
409    }
410
411    async fn list_workflow_ids(&self) -> Result<Vec<WorkflowId>, StoreError> {
412        crate::read::list_workflow_ids(self.connection()).await
413    }
414
415    async fn list_active(&self) -> Result<Vec<WorkflowId>, StoreError> {
416        crate::read::list_active(self.connection()).await
417    }
418
419    async fn list_paused(&self) -> Result<Vec<WorkflowId>, StoreError> {
420        crate::read::list_paused(self.connection()).await
421    }
422
423    async fn query(&self, filter: &WorkflowFilter) -> Result<Vec<WorkflowSummary>, StoreError> {
424        crate::read::query(self.connection(), filter).await
425    }
426
427    async fn schedule_timer(
428        &self,
429        workflow_id: &WorkflowId,
430        timer_id: &TimerId,
431        fire_at: DateTime<Utc>,
432    ) -> Result<(), StoreError> {
433        crate::timer::schedule_timer(self.connection(), workflow_id, timer_id, fire_at).await
434    }
435
436    async fn expired_timers(&self, as_of: DateTime<Utc>) -> Result<Vec<TimerEntry>, StoreError> {
437        crate::timer::expired_timers(self.connection(), as_of).await
438    }
439}
440
441#[cfg(test)]
442mod tests {
443    use std::path::PathBuf;
444    use std::sync::Arc;
445    use std::time::{SystemTime, UNIX_EPOCH};
446
447    use aion_store::{EventStore, StoreError};
448
449    use super::LibSqlStore;
450
451    #[test]
452    fn libsql_store_is_send_sync_static() {
453        fn assert_send_sync_static<T: Send + Sync + 'static>() {}
454
455        assert_send_sync_static::<LibSqlStore>();
456    }
457
458    #[tokio::test]
459    async fn open_creates_schema() -> Result<(), StoreError> {
460        let store = LibSqlStore::open(unique_temp_path("open-schema")).await?;
461
462        assert_schema_object(store.connection(), "table", "events").await?;
463        assert_schema_object(store.connection(), "table", "timers").await?;
464        assert_schema_object(store.connection(), "table", "visibility").await?;
465
466        Ok(())
467    }
468
469    #[tokio::test]
470    async fn store_can_be_used_as_event_store_trait_object() -> Result<(), StoreError> {
471        let store = LibSqlStore::open(unique_temp_path("trait-object")).await?;
472        let store: Arc<dyn EventStore> = Arc::new(store);
473
474        assert_eq!(Arc::strong_count(&store), 1);
475        Ok(())
476    }
477
478    #[tokio::test]
479    async fn connection_accessor_reuses_same_database_handle() -> Result<(), StoreError> {
480        let store = LibSqlStore::open(unique_temp_path("shared-handle")).await?;
481
482        store
483            .connection()
484            .execute(
485                "INSERT INTO timers (workflow_id, timer_id, fire_at) VALUES (?1, ?2, ?3)",
486                ("workflow-a", "timer-a", "2026-06-03T00:00:00Z"),
487            )
488            .await
489            .map_err(|error| crate::error::libsql_error(&error))?;
490
491        let count = timer_count(store.connection()).await?;
492        if count == 1 {
493            Ok(())
494        } else {
495            Err(StoreError::Backend(format!(
496                "expected one timer through shared connection, found {count}"
497            )))
498        }
499    }
500
501    async fn assert_schema_object(
502        conn: &libsql::Connection,
503        object_type: &str,
504        name: &str,
505    ) -> Result<(), StoreError> {
506        let mut rows = conn
507            .query(
508                "SELECT name FROM sqlite_master WHERE type = ?1 AND name = ?2",
509                (object_type, name),
510            )
511            .await
512            .map_err(|error| crate::error::libsql_error(&error))?;
513        let found = rows
514            .next()
515            .await
516            .map_err(|error| crate::error::libsql_error(&error))?
517            .is_some();
518
519        if found {
520            Ok(())
521        } else {
522            Err(StoreError::Backend(format!(
523                "schema object {object_type} {name} was not created"
524            )))
525        }
526    }
527
528    async fn timer_count(conn: &libsql::Connection) -> Result<i64, StoreError> {
529        let mut rows = conn
530            .query("SELECT COUNT(*) FROM timers", ())
531            .await
532            .map_err(|error| crate::error::libsql_error(&error))?;
533        let row = rows
534            .next()
535            .await
536            .map_err(|error| crate::error::libsql_error(&error))?
537            .ok_or_else(|| {
538                StoreError::Backend(String::from("timer count query returned no row"))
539            })?;
540
541        row.get(0)
542            .map_err(|error| crate::error::libsql_error(&error))
543    }
544
545    fn unique_temp_path(name: &str) -> PathBuf {
546        let nanos = SystemTime::now()
547            .duration_since(UNIX_EPOCH)
548            .map_or(0, |duration| duration.as_nanos());
549        std::env::temp_dir().join(format!(
550            "aion-store-libsql-store-{name}-{}-{nanos}.db",
551            std::process::id()
552        ))
553    }
554}
aion_store_libsql/store.rs

aion_store_libsql/
store.rs
