aion_store_libsql/
schema.rs

1//! Idempotent schema DDL for the libSQL event store.
2
3use aion_store::StoreError;
4
5/// Append-only workflow event table.
6pub const CREATE_EVENTS_TABLE: &str = "
7CREATE TABLE IF NOT EXISTS events (
8    workflow_id TEXT NOT NULL,
9    seq INTEGER NOT NULL,
10    event BLOB NOT NULL,
11    recorded_at TEXT NOT NULL,
12    event_kind TEXT NOT NULL,
13    is_queryable_event INTEGER NOT NULL,
14    workflow_type TEXT,
15    child_workflow_id TEXT,
16    PRIMARY KEY (workflow_id, seq)
17)";
18
19/// Event index supporting lifecycle projection scans and filter subqueries.
20pub const CREATE_EVENTS_PROJECTION_INDEX: &str = "
21CREATE INDEX IF NOT EXISTS idx_events_queryable_filter
22ON events (is_queryable_event, workflow_id, seq, event_kind, workflow_type, recorded_at, child_workflow_id)";
23
24/// Durable workflow timers table.
25pub const CREATE_TIMERS_TABLE: &str = "
26CREATE TABLE IF NOT EXISTS timers (
27    workflow_id TEXT NOT NULL,
28    timer_id TEXT NOT NULL,
29    fire_at TEXT NOT NULL,
30    PRIMARY KEY (workflow_id, timer_id)
31)";
32
33/// Timer index supporting due-timer range scans.
34pub const CREATE_TIMERS_FIRE_AT_INDEX: &str = "
35CREATE INDEX IF NOT EXISTS idx_timers_fire_at
36ON timers (fire_at)";
37
38/// Runtime-deployed package archives keyed by `(workflow_type, content_hash)`.
39pub const CREATE_PACKAGES_TABLE: &str = "
40CREATE TABLE IF NOT EXISTS packages (
41    workflow_type TEXT NOT NULL,
42    content_hash TEXT NOT NULL,
43    archive BLOB NOT NULL,
44    deployed_at TEXT NOT NULL,
45    PRIMARY KEY (workflow_type, content_hash)
46)";
47
48/// Per-workflow-type route pointer for new workflow starts.
49pub const CREATE_PACKAGE_ROUTES_TABLE: &str = "
50CREATE TABLE IF NOT EXISTS package_routes (
51    workflow_type TEXT PRIMARY KEY,
52    content_hash TEXT NOT NULL
53)";
54
55/// Workflow visibility projection table.
56pub const CREATE_VISIBILITY_TABLE: &str = "
57CREATE TABLE IF NOT EXISTS visibility (
58    workflow_id TEXT PRIMARY KEY,
59    run_id TEXT NOT NULL,
60    workflow_type TEXT NOT NULL,
61    status TEXT NOT NULL,
62    start_time TEXT NOT NULL,
63    close_time TEXT,
64    search_attributes TEXT NOT NULL CHECK (json_valid(search_attributes))
65)";
66
67/// Visibility index supporting workflow-type equality filters.
68pub const CREATE_VISIBILITY_WORKFLOW_TYPE_INDEX: &str = "
69CREATE INDEX IF NOT EXISTS idx_visibility_workflow_type
70ON visibility (workflow_type)";
71
72/// Visibility index supporting status equality filters.
73pub const CREATE_VISIBILITY_STATUS_INDEX: &str = "
74CREATE INDEX IF NOT EXISTS idx_visibility_status
75ON visibility (status)";
76
77/// Visibility index supporting start-time range filters and ordering.
78pub const CREATE_VISIBILITY_START_TIME_INDEX: &str = "
79CREATE INDEX IF NOT EXISTS idx_visibility_start_time
80ON visibility (start_time)";
81
82/// Visibility index supporting close-time range filters.
83pub const CREATE_VISIBILITY_CLOSE_TIME_INDEX: &str = "
84CREATE INDEX IF NOT EXISTS idx_visibility_close_time
85ON visibility (close_time)";
86
87/// Durable fan-out dispatch outbox.
88///
89/// `dispatch_key` (`"{workflow_id}:{ordinal}"`) is `UNIQUE`: it is the database-level idempotency
90/// guard, so a re-issued append of the same fan-out batch silently ignores the duplicate rows via
91/// `INSERT OR IGNORE`. `status` is one of `pending`/`claimed`/`done`/`failed`; `visible_after`
92/// fences retry backoff so a row is not re-claimed before its delay elapses; nullable
93/// `claimed_at` records the durable claim instant for live stale-claim reconciliation; nullable
94/// `run_id` records the concrete run that staged the row when known.
95pub const CREATE_OUTBOX_TABLE: &str = "
96CREATE TABLE IF NOT EXISTS outbox (
97    dispatch_key TEXT NOT NULL UNIQUE,
98    workflow_id TEXT NOT NULL,
99    ordinal INTEGER NOT NULL,
100    activity_type TEXT NOT NULL,
101    input BLOB NOT NULL,
102    status TEXT NOT NULL,
103    attempt INTEGER NOT NULL,
104    visible_after TEXT NOT NULL,
105    run_id TEXT,
106    claimed_at TEXT,
107    PRIMARY KEY (dispatch_key)
108)";
109
110/// Partial index over claimable rows, supporting the dispatcher's `status='pending'` claim scan
111/// ordered by `visible_after`. Restricting the index to pending rows keeps it small as completed
112/// rows accumulate.
113pub const CREATE_OUTBOX_PENDING_INDEX: &str = "
114CREATE INDEX IF NOT EXISTS idx_outbox_pending
115ON outbox (status, visible_after)
116WHERE status = 'pending'";
117
118/// Partial index over stale-claim reconciliation candidates.
119pub const CREATE_OUTBOX_CLAIMED_INDEX: &str = "
120CREATE INDEX IF NOT EXISTS idx_outbox_claimed_at
121ON outbox (status, claimed_at)
122WHERE status = 'claimed' AND claimed_at IS NOT NULL";
123
124const DDL_STATEMENTS: [&str; 13] = [
125    CREATE_EVENTS_TABLE,
126    CREATE_EVENTS_PROJECTION_INDEX,
127    CREATE_TIMERS_TABLE,
128    CREATE_TIMERS_FIRE_AT_INDEX,
129    CREATE_PACKAGES_TABLE,
130    CREATE_PACKAGE_ROUTES_TABLE,
131    CREATE_VISIBILITY_TABLE,
132    CREATE_VISIBILITY_WORKFLOW_TYPE_INDEX,
133    CREATE_VISIBILITY_STATUS_INDEX,
134    CREATE_VISIBILITY_START_TIME_INDEX,
135    CREATE_VISIBILITY_CLOSE_TIME_INDEX,
136    CREATE_OUTBOX_TABLE,
137    CREATE_OUTBOX_PENDING_INDEX,
138];
139
140/// Ensure the libSQL schema exists on a fresh or previously-created database.
141///
142/// # Errors
143///
144/// Returns `StoreError::Backend` when any idempotent DDL statement fails at the libSQL boundary.
145pub async fn ensure_schema(conn: &libsql::Connection) -> Result<(), StoreError> {
146    for statement in DDL_STATEMENTS {
147        conn.execute(statement, ())
148            .await
149            .map_err(|error| crate::error::libsql_error(&error))?;
150    }
151
152    ensure_outbox_claimed_at_column(conn).await?;
153    ensure_outbox_run_id_column(conn).await?;
154    ensure_outbox_namespace_column(conn).await?;
155    ensure_outbox_task_queue_column(conn).await?;
156    ensure_outbox_node_column(conn).await?;
157    conn.execute(CREATE_OUTBOX_CLAIMED_INDEX, ())
158        .await
159        .map_err(|error| crate::error::libsql_error(&error))?;
160
161    Ok(())
162}
163
164async fn ensure_outbox_claimed_at_column(conn: &libsql::Connection) -> Result<(), StoreError> {
165    if outbox_column_exists(conn, "claimed_at").await? {
166        return Ok(());
167    }
168
169    conn.execute("ALTER TABLE outbox ADD COLUMN claimed_at TEXT", ())
170        .await
171        .map(|_| ())
172        .map_err(|error| crate::error::libsql_error(&error))
173}
174
175async fn ensure_outbox_run_id_column(conn: &libsql::Connection) -> Result<(), StoreError> {
176    if outbox_column_exists(conn, "run_id").await? {
177        return Ok(());
178    }
179
180    conn.execute("ALTER TABLE outbox ADD COLUMN run_id TEXT", ())
181        .await
182        .map(|_| ())
183        .map_err(|error| crate::error::libsql_error(&error))
184}
185
186async fn ensure_outbox_namespace_column(conn: &libsql::Connection) -> Result<(), StoreError> {
187    if outbox_column_exists(conn, "namespace").await? {
188        return Ok(());
189    }
190
191    conn.execute("ALTER TABLE outbox ADD COLUMN namespace TEXT", ())
192        .await
193        .map(|_| ())
194        .map_err(|error| crate::error::libsql_error(&error))
195}
196
197async fn ensure_outbox_task_queue_column(conn: &libsql::Connection) -> Result<(), StoreError> {
198    if outbox_column_exists(conn, "task_queue").await? {
199        return Ok(());
200    }
201
202    conn.execute("ALTER TABLE outbox ADD COLUMN task_queue TEXT", ())
203        .await
204        .map(|_| ())
205        .map_err(|error| crate::error::libsql_error(&error))
206}
207
208async fn ensure_outbox_node_column(conn: &libsql::Connection) -> Result<(), StoreError> {
209    if outbox_column_exists(conn, "node").await? {
210        return Ok(());
211    }
212
213    conn.execute("ALTER TABLE outbox ADD COLUMN node TEXT", ())
214        .await
215        .map(|_| ())
216        .map_err(|error| crate::error::libsql_error(&error))
217}
218
219async fn outbox_column_exists(
220    conn: &libsql::Connection,
221    column_name: &str,
222) -> Result<bool, StoreError> {
223    let mut rows = conn
224        .query("PRAGMA table_info(outbox)", ())
225        .await
226        .map_err(|error| crate::error::libsql_error(&error))?;
227
228    while let Some(row) = rows
229        .next()
230        .await
231        .map_err(|error| crate::error::libsql_error(&error))?
232    {
233        let name: String = row
234            .get(1)
235            .map_err(|error| crate::error::libsql_error(&error))?;
236        if name == column_name {
237            return Ok(true);
238        }
239    }
240
241    Ok(false)
242}
243
244#[cfg(test)]
245mod tests {
246    use std::path::PathBuf;
247    use std::time::{SystemTime, UNIX_EPOCH};
248
249    use aion_store::StoreError;
250
251    use super::ensure_schema;
252    use crate::config::{LibSqlConfig, LibSqlMode};
253    use crate::connection::open_connection;
254
255    #[tokio::test]
256    async fn ensure_schema_is_idempotent() -> Result<(), StoreError> {
257        let conn = open_test_connection("idempotent").await?;
258
259        ensure_schema(&conn).await?;
260        ensure_schema(&conn).await?;
261
262        Ok(())
263    }
264
265    #[tokio::test]
266    async fn ensure_schema_creates_tables_and_indexes() -> Result<(), StoreError> {
267        let conn = open_test_connection("objects").await?;
268
269        ensure_schema(&conn).await?;
270
271        assert_schema_object(&conn, "table", "events").await?;
272        assert_schema_object(&conn, "index", "sqlite_autoindex_events_1").await?;
273        assert_schema_object(&conn, "index", "idx_events_queryable_filter").await?;
274        assert_schema_object(&conn, "table", "timers").await?;
275        assert_schema_object(&conn, "index", "sqlite_autoindex_timers_1").await?;
276        assert_schema_object(&conn, "index", "idx_timers_fire_at").await?;
277        assert_schema_object(&conn, "table", "packages").await?;
278        assert_schema_object(&conn, "index", "sqlite_autoindex_packages_1").await?;
279        assert_schema_object(&conn, "table", "package_routes").await?;
280        assert_schema_object(&conn, "index", "sqlite_autoindex_package_routes_1").await?;
281        assert_schema_object(&conn, "table", "visibility").await?;
282        assert_schema_object(&conn, "index", "sqlite_autoindex_visibility_1").await?;
283        assert_schema_object(&conn, "index", "idx_visibility_workflow_type").await?;
284        assert_schema_object(&conn, "index", "idx_visibility_status").await?;
285        assert_schema_object(&conn, "index", "idx_visibility_start_time").await?;
286        assert_schema_object(&conn, "index", "idx_visibility_close_time").await?;
287        assert_schema_object(&conn, "table", "outbox").await?;
288        assert_schema_object(&conn, "index", "idx_outbox_pending").await?;
289        assert_schema_object(&conn, "index", "idx_outbox_claimed_at").await?;
290
291        Ok(())
292    }
293
294    async fn open_test_connection(name: &str) -> Result<libsql::Connection, StoreError> {
295        let config = LibSqlConfig {
296            mode: LibSqlMode::Embedded {
297                path: unique_temp_path(name),
298            },
299            journal_mode: None,
300            synchronous: None,
301            sync_interval_seconds: None,
302        };
303
304        open_connection(&config)
305            .await
306            .map(|opened| opened.connection)
307    }
308
309    async fn assert_schema_object(
310        conn: &libsql::Connection,
311        object_type: &str,
312        name: &str,
313    ) -> Result<(), StoreError> {
314        let mut rows = conn
315            .query(
316                "SELECT name FROM sqlite_master WHERE type = ?1 AND name = ?2",
317                (object_type, name),
318            )
319            .await
320            .map_err(|error| crate::error::libsql_error(&error))?;
321        let found = rows
322            .next()
323            .await
324            .map_err(|error| crate::error::libsql_error(&error))?
325            .is_some();
326
327        if found {
328            Ok(())
329        } else {
330            Err(StoreError::Backend(format!(
331                "schema object {object_type} {name} was not created"
332            )))
333        }
334    }
335
336    fn unique_temp_path(name: &str) -> PathBuf {
337        let nanos = SystemTime::now()
338            .duration_since(UNIX_EPOCH)
339            .map_or(0, |duration| duration.as_nanos());
340        std::env::temp_dir().join(format!(
341            "aion-store-libsql-schema-{name}-{}-{nanos}.db",
342            std::process::id()
343        ))
344    }
345}
aion_store_libsql/schema.rs

aion_store_libsql/
schema.rs
