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    failed_step TEXT,
65    failure_reason TEXT,
66    search_attributes TEXT NOT NULL CHECK (json_valid(search_attributes))
67)";
68
69/// Visibility index supporting workflow-type equality filters.
70pub const CREATE_VISIBILITY_WORKFLOW_TYPE_INDEX: &str = "
71CREATE INDEX IF NOT EXISTS idx_visibility_workflow_type
72ON visibility (workflow_type)";
73
74/// Visibility index supporting status equality filters.
75pub const CREATE_VISIBILITY_STATUS_INDEX: &str = "
76CREATE INDEX IF NOT EXISTS idx_visibility_status
77ON visibility (status)";
78
79/// Visibility index supporting start-time range filters and ordering.
80pub const CREATE_VISIBILITY_START_TIME_INDEX: &str = "
81CREATE INDEX IF NOT EXISTS idx_visibility_start_time
82ON visibility (start_time)";
83
84/// Visibility index supporting close-time range filters.
85pub const CREATE_VISIBILITY_CLOSE_TIME_INDEX: &str = "
86CREATE INDEX IF NOT EXISTS idx_visibility_close_time
87ON visibility (close_time)";
88
89/// Durable minted-on-use namespace registry (Control-Plane Phase 1).
90///
91/// One row per namespace keyed by `name`. `created_at`/`last_seen` are RFC 3339
92/// text (matching the package/timer/outbox encodings); `record` holds the opaque
93/// [`NamespaceRecord`](aion_store::NamespaceRecord) codec bytes that are the
94/// store's truth (only decoded to satisfy `list`). The single-node libSQL path
95/// upserts locally with no quorum to reach.
96pub const CREATE_NAMESPACES_TABLE: &str = "
97CREATE TABLE IF NOT EXISTS namespaces (
98    name TEXT PRIMARY KEY,
99    created_at TEXT NOT NULL,
100    last_seen TEXT NOT NULL,
101    record BLOB NOT NULL
102)";
103
104/// Namespace index supporting the `created_at` ASC (tie-break `name`) list ordering.
105pub const CREATE_NAMESPACES_CREATED_AT_INDEX: &str = "
106CREATE INDEX IF NOT EXISTS idx_namespaces_created_at
107ON namespaces (created_at, name)";
108
109/// Durable fan-out dispatch outbox.
110///
111/// `dispatch_key` (`"{workflow_id}:{ordinal}"`) is `UNIQUE`: it is the database-level idempotency
112/// guard, so a re-issued append of the same fan-out batch silently ignores the duplicate rows via
113/// `INSERT OR IGNORE`. `status` is one of `pending`/`claimed`/`done`/`failed`; `visible_after`
114/// fences retry backoff so a row is not re-claimed before its delay elapses; nullable
115/// `claimed_at` records the durable claim instant for live stale-claim reconciliation; nullable
116/// `run_id` records the concrete run that staged the row when known.
117pub const CREATE_OUTBOX_TABLE: &str = "
118CREATE TABLE IF NOT EXISTS outbox (
119    dispatch_key TEXT NOT NULL UNIQUE,
120    workflow_id TEXT NOT NULL,
121    ordinal INTEGER NOT NULL,
122    activity_type TEXT NOT NULL,
123    input BLOB NOT NULL,
124    status TEXT NOT NULL,
125    attempt INTEGER NOT NULL,
126    visible_after TEXT NOT NULL,
127    run_id TEXT,
128    claimed_at TEXT,
129    PRIMARY KEY (dispatch_key)
130)";
131
132/// Partial index over claimable rows, supporting the dispatcher's `status='pending'` claim scan
133/// ordered by `visible_after`. Restricting the index to pending rows keeps it small as completed
134/// rows accumulate.
135pub const CREATE_OUTBOX_PENDING_INDEX: &str = "
136CREATE INDEX IF NOT EXISTS idx_outbox_pending
137ON outbox (status, visible_after)
138WHERE status = 'pending'";
139
140/// Partial index over stale-claim reconciliation candidates.
141pub const CREATE_OUTBOX_CLAIMED_INDEX: &str = "
142CREATE INDEX IF NOT EXISTS idx_outbox_claimed_at
143ON outbox (status, claimed_at)
144WHERE status = 'claimed' AND claimed_at IS NOT NULL";
145
146const DDL_STATEMENTS: [&str; 15] = [
147    CREATE_EVENTS_TABLE,
148    CREATE_EVENTS_PROJECTION_INDEX,
149    CREATE_TIMERS_TABLE,
150    CREATE_TIMERS_FIRE_AT_INDEX,
151    CREATE_PACKAGES_TABLE,
152    CREATE_PACKAGE_ROUTES_TABLE,
153    CREATE_VISIBILITY_TABLE,
154    CREATE_VISIBILITY_WORKFLOW_TYPE_INDEX,
155    CREATE_VISIBILITY_STATUS_INDEX,
156    CREATE_VISIBILITY_START_TIME_INDEX,
157    CREATE_VISIBILITY_CLOSE_TIME_INDEX,
158    CREATE_OUTBOX_TABLE,
159    CREATE_OUTBOX_PENDING_INDEX,
160    CREATE_NAMESPACES_TABLE,
161    CREATE_NAMESPACES_CREATED_AT_INDEX,
162];
163
164/// Ensure the libSQL schema exists on a fresh or previously-created database.
165///
166/// # Errors
167///
168/// Returns `StoreError::Backend` when any idempotent DDL statement fails at the libSQL boundary.
169pub async fn ensure_schema(conn: &libsql::Connection) -> Result<(), StoreError> {
170    for statement in DDL_STATEMENTS {
171        conn.execute(statement, ())
172            .await
173            .map_err(|error| crate::error::libsql_error(&error))?;
174    }
175
176    ensure_outbox_claimed_at_column(conn).await?;
177    ensure_outbox_run_id_column(conn).await?;
178    ensure_outbox_namespace_column(conn).await?;
179    ensure_outbox_task_queue_column(conn).await?;
180    ensure_outbox_node_column(conn).await?;
181    conn.execute(CREATE_OUTBOX_CLAIMED_INDEX, ())
182        .await
183        .map_err(|error| crate::error::libsql_error(&error))?;
184
185    Ok(())
186}
187
188async fn ensure_outbox_claimed_at_column(conn: &libsql::Connection) -> Result<(), StoreError> {
189    if outbox_column_exists(conn, "claimed_at").await? {
190        return Ok(());
191    }
192
193    conn.execute("ALTER TABLE outbox ADD COLUMN claimed_at TEXT", ())
194        .await
195        .map(|_| ())
196        .map_err(|error| crate::error::libsql_error(&error))
197}
198
199async fn ensure_outbox_run_id_column(conn: &libsql::Connection) -> Result<(), StoreError> {
200    if outbox_column_exists(conn, "run_id").await? {
201        return Ok(());
202    }
203
204    conn.execute("ALTER TABLE outbox ADD COLUMN run_id TEXT", ())
205        .await
206        .map(|_| ())
207        .map_err(|error| crate::error::libsql_error(&error))
208}
209
210async fn ensure_outbox_namespace_column(conn: &libsql::Connection) -> Result<(), StoreError> {
211    if outbox_column_exists(conn, "namespace").await? {
212        return Ok(());
213    }
214
215    conn.execute("ALTER TABLE outbox ADD COLUMN namespace TEXT", ())
216        .await
217        .map(|_| ())
218        .map_err(|error| crate::error::libsql_error(&error))
219}
220
221async fn ensure_outbox_task_queue_column(conn: &libsql::Connection) -> Result<(), StoreError> {
222    if outbox_column_exists(conn, "task_queue").await? {
223        return Ok(());
224    }
225
226    conn.execute("ALTER TABLE outbox ADD COLUMN task_queue TEXT", ())
227        .await
228        .map(|_| ())
229        .map_err(|error| crate::error::libsql_error(&error))
230}
231
232async fn ensure_outbox_node_column(conn: &libsql::Connection) -> Result<(), StoreError> {
233    if outbox_column_exists(conn, "node").await? {
234        return Ok(());
235    }
236
237    conn.execute("ALTER TABLE outbox ADD COLUMN node TEXT", ())
238        .await
239        .map(|_| ())
240        .map_err(|error| crate::error::libsql_error(&error))
241}
242
243async fn outbox_column_exists(
244    conn: &libsql::Connection,
245    column_name: &str,
246) -> Result<bool, StoreError> {
247    let mut rows = conn
248        .query("PRAGMA table_info(outbox)", ())
249        .await
250        .map_err(|error| crate::error::libsql_error(&error))?;
251
252    while let Some(row) = rows
253        .next()
254        .await
255        .map_err(|error| crate::error::libsql_error(&error))?
256    {
257        let name: String = row
258            .get(1)
259            .map_err(|error| crate::error::libsql_error(&error))?;
260        if name == column_name {
261            return Ok(true);
262        }
263    }
264
265    Ok(false)
266}
267
268#[cfg(test)]
269mod tests {
270    use std::path::PathBuf;
271    use std::time::{SystemTime, UNIX_EPOCH};
272
273    use aion_store::StoreError;
274
275    use super::ensure_schema;
276    use crate::config::{LibSqlConfig, LibSqlMode};
277    use crate::connection::open_connection;
278
279    #[tokio::test]
280    async fn ensure_schema_is_idempotent() -> Result<(), StoreError> {
281        let conn = open_test_connection("idempotent").await?;
282
283        ensure_schema(&conn).await?;
284        ensure_schema(&conn).await?;
285
286        Ok(())
287    }
288
289    #[tokio::test]
290    async fn ensure_schema_creates_tables_and_indexes() -> Result<(), StoreError> {
291        let conn = open_test_connection("objects").await?;
292
293        ensure_schema(&conn).await?;
294
295        assert_schema_object(&conn, "table", "events").await?;
296        assert_schema_object(&conn, "index", "sqlite_autoindex_events_1").await?;
297        assert_schema_object(&conn, "index", "idx_events_queryable_filter").await?;
298        assert_schema_object(&conn, "table", "timers").await?;
299        assert_schema_object(&conn, "index", "sqlite_autoindex_timers_1").await?;
300        assert_schema_object(&conn, "index", "idx_timers_fire_at").await?;
301        assert_schema_object(&conn, "table", "packages").await?;
302        assert_schema_object(&conn, "index", "sqlite_autoindex_packages_1").await?;
303        assert_schema_object(&conn, "table", "package_routes").await?;
304        assert_schema_object(&conn, "index", "sqlite_autoindex_package_routes_1").await?;
305        assert_schema_object(&conn, "table", "visibility").await?;
306        assert_schema_object(&conn, "index", "sqlite_autoindex_visibility_1").await?;
307        assert_schema_object(&conn, "index", "idx_visibility_workflow_type").await?;
308        assert_schema_object(&conn, "index", "idx_visibility_status").await?;
309        assert_schema_object(&conn, "index", "idx_visibility_start_time").await?;
310        assert_schema_object(&conn, "index", "idx_visibility_close_time").await?;
311        assert_schema_object(&conn, "table", "outbox").await?;
312        assert_schema_object(&conn, "index", "idx_outbox_pending").await?;
313        assert_schema_object(&conn, "index", "idx_outbox_claimed_at").await?;
314        assert_schema_object(&conn, "table", "namespaces").await?;
315        assert_schema_object(&conn, "index", "sqlite_autoindex_namespaces_1").await?;
316        assert_schema_object(&conn, "index", "idx_namespaces_created_at").await?;
317
318        Ok(())
319    }
320
321    async fn open_test_connection(name: &str) -> Result<libsql::Connection, StoreError> {
322        let config = LibSqlConfig {
323            mode: LibSqlMode::Embedded {
324                path: unique_temp_path(name),
325            },
326            journal_mode: None,
327            synchronous: None,
328            sync_interval_seconds: None,
329        };
330
331        open_connection(&config)
332            .await
333            .map(|opened| opened.connection)
334    }
335
336    async fn assert_schema_object(
337        conn: &libsql::Connection,
338        object_type: &str,
339        name: &str,
340    ) -> Result<(), StoreError> {
341        let mut rows = conn
342            .query(
343                "SELECT name FROM sqlite_master WHERE type = ?1 AND name = ?2",
344                (object_type, name),
345            )
346            .await
347            .map_err(|error| crate::error::libsql_error(&error))?;
348        let found = rows
349            .next()
350            .await
351            .map_err(|error| crate::error::libsql_error(&error))?
352            .is_some();
353
354        if found {
355            Ok(())
356        } else {
357            Err(StoreError::Backend(format!(
358                "schema object {object_type} {name} was not created"
359            )))
360        }
361    }
362
363    fn unique_temp_path(name: &str) -> PathBuf {
364        let nanos = SystemTime::now()
365            .duration_since(UNIX_EPOCH)
366            .map_or(0, |duration| duration.as_nanos());
367        std::env::temp_dir().join(format!(
368            "aion-store-libsql-schema-{name}-{}-{nanos}.db",
369            std::process::id()
370        ))
371    }
372}
aion_store_libsql/schema.rs

aion_store_libsql/
schema.rs
