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

aion_store_libsql/
store.rs
