zeph_memory/sqlite/
mod.rs

1// SPDX-FileCopyrightText: 2026 Andrei G <bug-ops>
2// SPDX-License-Identifier: MIT OR Apache-2.0
3
4mod acp_sessions;
5pub mod compression_guidelines;
6pub mod corrections;
7#[cfg(feature = "experiments")]
8pub mod experiments;
9pub mod graph_store;
10mod history;
11pub(crate) mod messages;
12pub mod overflow;
13pub mod preferences;
14mod skills;
15mod summaries;
16mod trust;
17
18use sqlx::SqlitePool;
19use sqlx::sqlite::{SqliteConnectOptions, SqlitePoolOptions};
20use std::str::FromStr;
21use std::time::Duration;
22
23use crate::error::MemoryError;
24
25pub use acp_sessions::{AcpSessionEvent, AcpSessionInfo};
26pub use messages::role_str;
27pub use skills::{SkillMetricsRow, SkillUsageRow, SkillVersionRow};
28pub use trust::{SkillTrustRow, SourceKind};
29
30#[derive(Debug, Clone)]
31pub struct SqliteStore {
32    pool: SqlitePool,
33}
34
35impl SqliteStore {
36    const DEFAULT_BUSY_TIMEOUT: Duration = Duration::from_secs(5);
37
38    /// Open (or create) the `SQLite` database and run migrations.
39    ///
40    /// Enables foreign key constraints at connection level so that
41    /// `ON DELETE CASCADE` and other FK rules are enforced.
42    ///
43    /// # Errors
44    ///
45    /// Returns an error if the database cannot be opened or migrations fail.
46    pub async fn new(path: &str) -> Result<Self, MemoryError> {
47        Self::with_pool_size(path, 5).await
48    }
49
50    /// Open (or create) the `SQLite` database with a configurable connection pool size.
51    ///
52    /// # Errors
53    ///
54    /// Returns an error if the database cannot be opened or migrations fail.
55    pub async fn with_pool_size(path: &str, pool_size: u32) -> Result<Self, MemoryError> {
56        let url = if path == ":memory:" {
57            "sqlite::memory:".to_string()
58        } else {
59            if let Some(parent) = std::path::Path::new(path).parent()
60                && !parent.as_os_str().is_empty()
61            {
62                std::fs::create_dir_all(parent)?;
63            }
64            format!("sqlite:{path}?mode=rwc")
65        };
66
67        let opts = SqliteConnectOptions::from_str(&url)?
68            .create_if_missing(true)
69            .foreign_keys(true)
70            .busy_timeout(Self::DEFAULT_BUSY_TIMEOUT)
71            .journal_mode(sqlx::sqlite::SqliteJournalMode::Wal)
72            .synchronous(sqlx::sqlite::SqliteSynchronous::Normal);
73
74        let pool = SqlitePoolOptions::new()
75            .max_connections(pool_size)
76            .connect_with(opts)
77            .await?;
78
79        sqlx::migrate!("./migrations").run(&pool).await?;
80
81        if path != ":memory:" {
82            sqlx::query("PRAGMA wal_checkpoint(PASSIVE)")
83                .execute(&pool)
84                .await?;
85        }
86
87        Ok(Self { pool })
88    }
89
90    /// Expose the underlying pool for shared access by other stores.
91    #[must_use]
92    pub fn pool(&self) -> &SqlitePool {
93        &self.pool
94    }
95
96    /// Run all migrations on the given pool.
97    ///
98    /// # Errors
99    ///
100    /// Returns an error if any migration fails.
101    pub async fn run_migrations(pool: &SqlitePool) -> Result<(), MemoryError> {
102        sqlx::migrate!("./migrations").run(pool).await?;
103        Ok(())
104    }
105}
106
107#[cfg(test)]
108mod tests {
109    use super::*;
110    use tempfile::NamedTempFile;
111
112    #[tokio::test]
113    async fn wal_journal_mode_enabled_on_file_db() {
114        let file = NamedTempFile::new().expect("tempfile");
115        let path = file.path().to_str().expect("valid path");
116
117        let store = SqliteStore::new(path).await.expect("SqliteStore::new");
118
119        let mode: String = sqlx::query_scalar("PRAGMA journal_mode")
120            .fetch_one(store.pool())
121            .await
122            .expect("PRAGMA query");
123
124        assert_eq!(mode, "wal", "expected WAL journal mode, got: {mode}");
125    }
126
127    #[tokio::test]
128    async fn busy_timeout_enabled_on_file_db() {
129        let file = NamedTempFile::new().expect("tempfile");
130        let path = file.path().to_str().expect("valid path");
131
132        let store = SqliteStore::new(path).await.expect("SqliteStore::new");
133
134        let timeout_ms: i64 = sqlx::query_scalar("PRAGMA busy_timeout")
135            .fetch_one(store.pool())
136            .await
137            .expect("PRAGMA query");
138
139        assert_eq!(
140            timeout_ms,
141            i64::try_from(SqliteStore::DEFAULT_BUSY_TIMEOUT.as_millis()).unwrap(),
142            "expected busy_timeout pragma to match configured timeout"
143        );
144    }
145
146    #[tokio::test]
147    async fn creates_parent_dirs() {
148        let dir = tempfile::tempdir().expect("tempdir");
149        let deep = dir.path().join("a/b/c/zeph.db");
150        let path = deep.to_str().expect("valid path");
151        let _store = SqliteStore::new(path).await.expect("SqliteStore::new");
152        assert!(deep.exists(), "database file should exist");
153    }
154
155    #[tokio::test]
156    async fn with_pool_size_accepts_custom_size() {
157        let store = SqliteStore::with_pool_size(":memory:", 2)
158            .await
159            .expect("with_pool_size");
160        // Verify the store is operational with the custom pool size.
161        let _cid = store
162            .create_conversation()
163            .await
164            .expect("create_conversation");
165    }
166}
zeph_memory/sqlite/mod.rs

zeph_memory/sqlite/
mod.rs
