attune_sqlite/

backend.rs

use std::{
    collections::HashMap,
    path::{Path, PathBuf},
    sync::Mutex,
    thread::{self, JoinHandle},
    time::{Duration, SystemTime, UNIX_EPOCH},
};

use attune_core::{BackendError, StorageBackend, StoredValue};
use crossbeam_channel::{Receiver, RecvTimeoutError, Sender, unbounded};
use rusqlite::{Connection, params};

const POLL_INTERVAL: Duration = Duration::from_millis(1000);

#[derive(Clone, Debug, Eq, PartialEq)]
pub struct SqliteOptions {
    pub cross_process: bool,
    pub poll_interval: Duration,
    pub journal_mode: SqliteJournalMode,
    pub busy_timeout: Duration,
    pub synchronous: SqliteSynchronous,
}

impl Default for SqliteOptions {
    fn default() -> Self {
        Self {
            cross_process: true,
            poll_interval: POLL_INTERVAL,
            journal_mode: SqliteJournalMode::Wal,
            busy_timeout: Duration::from_millis(5000),
            synchronous: SqliteSynchronous::Normal,
        }
    }
}

#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub enum SqliteJournalMode {
    Wal,
    Delete,
    Truncate,
    Persist,
    Memory,
    Off,
}

impl SqliteJournalMode {
    fn as_pragma(self) -> &'static str {
        match self {
            Self::Wal => "WAL",
            Self::Delete => "DELETE",
            Self::Truncate => "TRUNCATE",
            Self::Persist => "PERSIST",
            Self::Memory => "MEMORY",
            Self::Off => "OFF",
        }
    }
}

#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub enum SqliteSynchronous {
    Off,
    Normal,
    Full,
    Extra,
}

impl SqliteSynchronous {
    fn as_pragma(self) -> &'static str {
        match self {
            Self::Off => "OFF",
            Self::Normal => "NORMAL",
            Self::Full => "FULL",
            Self::Extra => "EXTRA",
        }
    }
}

pub struct SqliteBackend {
    conn: Mutex<Connection>,
    commits_rx: Option<Receiver<()>>,
    shutdown_tx: Option<Sender<()>>,
    poll_thread: Option<JoinHandle<()>>,
}

impl SqliteBackend {
    /// Opens a connection to the SQLite database.
    ///
    /// Uses [`SqliteOptions::default`] for SQLite pragmas and cross-process
    /// change detection.
    ///
    /// ## Errors
    ///
    /// Returns [`BackendError::Open`] when the database cannot be opened,
    /// SQLite pragmas cannot be applied, or the settings table cannot be
    /// created.
    pub fn open(path: impl AsRef<Path>) -> Result<Self, BackendError> {
        Self::open_with_options(path, SqliteOptions::default())
    }

    /// Opens a connection to the SQLite database with explicit backend options.
    ///
    /// Applies the configured SQLite pragmas, creates the settings table, and
    /// starts the cross-process watcher when `options.cross_process` is `true`.
    ///
    /// ## Errors
    ///
    /// Returns [`BackendError::Open`] when the database cannot be opened,
    /// SQLite pragmas cannot be applied, or the settings table cannot be
    /// created.
    pub fn open_with_options(
        path: impl AsRef<Path>,
        options: SqliteOptions,
    ) -> Result<Self, BackendError> {
        let path = path.as_ref().to_path_buf();

        // 1. Create the SQLite connection.
        let conn =
            rusqlite::Connection::open(&path).map_err(|e| BackendError::Open(e.to_string()))?;

        // 2. Set PRAGMAs.
        let pragmas = format!(
            "PRAGMA journal_mode = {};\
            PRAGMA busy_timeout = {};\
            PRAGMA synchronous = {};\
            PRAGMA foreign_keys = ON;",
            options.journal_mode.as_pragma(),
            options.busy_timeout.as_millis(),
            options.synchronous.as_pragma(),
        );
        conn.execute_batch(&pragmas)
            .map_err(|e| BackendError::Open(e.to_string()))?;

        // 3. Create settings table.
        let settings_table_sql = "CREATE TABLE IF NOT EXISTS settings (
            key TEXT PRIMARY KEY NOT NULL,
            value TEXT NOT NULL,
            updated_at INTEGER NOT NULL
        )";
        conn.execute(settings_table_sql, [])
            .map_err(|e| BackendError::Open(e.to_string()))?;

        // 4. Setup optional sidecar thread for cross-process change detection.
        let (commits_rx, shutdown_tx, poll_thread) = if options.cross_process {
            let (commits_tx, commits_rx) = unbounded::<()>();
            let (shutdown_tx, shutdown_rx) = unbounded::<()>();
            let sidecar_path = path.clone();
            let poll_interval = options.poll_interval;
            let poll_thread = thread::spawn(move || {
                polling_loop(sidecar_path, commits_tx, shutdown_rx, poll_interval);
            });

            (Some(commits_rx), Some(shutdown_tx), Some(poll_thread))
        } else {
            (None, None, None)
        };

        Ok(SqliteBackend {
            conn: Mutex::new(conn),
            commits_rx,
            shutdown_tx,
            poll_thread,
        })
    }
}

/// Run the cross-process change-detection loop on a sidecar SQLite connection.
///
/// Opens its own [`Connection`] to `path`, separate from the writer connection,
/// then polls `PRAGMA data_version` every `poll_interval`. Whenever the
/// version counter ticks (signalling that some connection committed to the
/// database), pushes `()` onto `commits_tx`. Own-process commits also tick the
/// counter; the diff loop in `attune-core` is responsible for dedup.
///
/// Returns silently, never panics, and never returns a `Result`. The loop exits
/// when any of these conditions occur:
///
/// - A shutdown value is received on `shutdown_rx` (clean shutdown requested
///   by the owning [`SqliteBackend`] being dropped).
/// - `shutdown_rx` becomes disconnected because its sender was dropped (also
///   indicates the backend has been dropped).
/// - `commits_tx` becomes disconnected because no receiver is listening for
///   change signals.
/// - Opening the sidecar connection fails, or any `PRAGMA data_version` query
///   fails. Cross-process detection degrades gracefully on storage errors
///   rather than propagating.
fn polling_loop(
    path: PathBuf,
    commits_tx: Sender<()>,
    shutdown_rx: Receiver<()>,
    poll_interval: Duration,
) {
    // 1. Open the sidecar connection.
    let sidecar_conn = match Connection::open(&path) {
        Ok(c) => c,
        Err(_) => return,
    };

    // 2. Get the initial data version.
    let mut last_version: i64 =
        match sidecar_conn.query_row("PRAGMA data_version", [], |row| row.get(0)) {
            Ok(v) => v,
            Err(_) => return,
        };

    loop {
        // Wait up to the poll interval for a shutdown signal.
        match shutdown_rx.recv_timeout(poll_interval) {
            Ok(()) => return,                              // Shutdown requested.
            Err(RecvTimeoutError::Disconnected) => return, // Sender dropped. (backend dropped)
            Err(RecvTimeoutError::Timeout) => {}           // Time to poll.
        }

        // Check the current data version.
        let version: i64 = match sidecar_conn.query_row("PRAGMA data_version", [], |row| row.get(0))
        {
            Ok(v) => v,
            Err(_) => return,
        };

        if version != last_version {
            last_version = version;

            // If the send fails, no one is listening. Silently exit.
            if commits_tx.send(()).is_err() {
                return;
            }
        }
    }
}

impl StorageBackend for SqliteBackend {
    fn load_all(&self) -> Result<HashMap<String, StoredValue>, BackendError> {
        // 1. Obtain lock to the DB connection.
        let conn = self.conn.lock().unwrap();

        // 2. Query DB for settings and deserialize values into StoredValue.
        let sql = "SELECT key, value FROM settings";
        let mut stmt = conn
            .prepare(sql)
            .map_err(|e| BackendError::Read(e.to_string()))?;
        let rows = stmt
            .query_map([], |row| {
                let key = row.get(0)?;
                let raw = row.get(1)?;
                Ok((key, StoredValue::from_raw(raw)))
            })
            .map_err(|e| BackendError::Read(e.to_string()))?;

        // 3. Add deserialized values to HashMap.
        let mut result = HashMap::new();
        for row in rows {
            let (k, v) = row.map_err(|e| BackendError::Read(e.to_string()))?;
            result.insert(k, v);
        }

        Ok(result)
    }

    fn set(&self, key: &str, value: &StoredValue) -> Result<(), BackendError> {
        // 1. Obtain lock to the DB connection.
        let conn = self.conn.lock().unwrap();

        // 2. Write setting to DB.
        let now = SystemTime::now()
            .duration_since(UNIX_EPOCH)
            .unwrap_or_default()
            .as_secs() as i64;
        let sql = "INSERT OR REPLACE INTO settings (key, value, updated_at) VALUES (?, ?, ?)";
        conn.execute(sql, params![key, value.as_str(), now])
            .map_err(|e| BackendError::Write(e.to_string()))?;

        Ok(())
    }

    fn delete(&self, key: &str) -> Result<(), BackendError> {
        // 1. Obtain lock to the DB connection.
        let conn = self.conn.lock().unwrap();

        // 2. Delete setting from DB.
        let sql = "DELETE FROM settings WHERE key = ?";
        conn.execute(sql, params![key])
            .map_err(|e| BackendError::Write(e.to_string()))?;

        Ok(())
    }

    fn watch_changes(&self) -> Option<Receiver<()>> {
        self.commits_rx.clone()
    }
}

impl Drop for SqliteBackend {
    fn drop(&mut self) {
        // Signal the polling thread to exit.
        if let Some(shutdown_tx) = &self.shutdown_tx {
            let _ = shutdown_tx.send(());
        }

        // Take the join handle out of Option and move it into `.join()`.
        if let Some(handle) = self.poll_thread.take() {
            let _ = handle.join();
        }
    }
}

#[cfg(test)]
mod test {
    use super::*;
    use rusqlite::OptionalExtension;

    #[test]
    fn test_open_correctly_inits_sqlite_db() {
        let sqlite_be = SqliteBackend::open(":memory:").unwrap();
        let conn = sqlite_be.conn.lock().unwrap();

        // Assert that the settings table is created.
        let mut stmt = conn
            .prepare("SELECT name FROM sqlite_master WHERE type='table' AND name='settings'")
            .unwrap();
        let result: Option<String> = stmt.query_row([], |row| row.get(0)).optional().unwrap();
        assert!(result.is_some());
        let result = result.unwrap();
        assert_eq!(result, "settings");
    }

    #[test]
    fn test_load_all_returns_a_hashmap_with_no_values() {
        let sqlite_be = SqliteBackend::open(":memory:").unwrap();
        let stored_values = sqlite_be.load_all().unwrap();
        assert_eq!(stored_values.len(), 0)
    }

    #[test]
    fn test_set_successfully_writes_a_setting_to_the_db() {
        let key = "theme";
        let sqlite_be = SqliteBackend::open(":memory:").unwrap();
        let sv = StoredValue::encode(&"dark").unwrap();
        sqlite_be.set(&key, &sv).unwrap();

        let stored_values = sqlite_be.load_all().unwrap();
        let loaded = stored_values.get::<str>(&key).unwrap();
        assert_eq!(stored_values.len(), 1);
        assert_eq!(sv.as_str(), loaded.as_str())
    }

    #[test]
    fn test_delete_successfully_removes_a_setting_from_the_db() {
        // 1. Write setting.
        let key = "theme";
        let sqlite_be = SqliteBackend::open(":memory:").unwrap();
        let sv = StoredValue::encode(&"dark").unwrap();
        sqlite_be.set(&key, &sv).unwrap();

        // 2. Ensure setting persists.
        let stored_values = sqlite_be.load_all().unwrap();
        assert_eq!(stored_values.len(), 1);

        // 3. Remove setting and ensure it's no longer in the DB.
        sqlite_be.delete(&key).unwrap();
        let stored_values = sqlite_be.load_all().unwrap();
        assert_eq!(stored_values.len(), 0);
    }

    #[test]
    fn test_watch_changes_signals_on_external_commit() {
        let tmp = tempfile::NamedTempFile::new().unwrap();
        let path = tmp.path().to_path_buf();

        // Open the backend
        let backend = SqliteBackend::open(&path).unwrap();
        let rx = backend
            .watch_changes()
            .expect("polling thread should be running");

        // Write through a second backend connection. This simulates an external process.
        let other = rusqlite::Connection::open(&path).unwrap();
        other
            .execute(
                "INSERT INTO settings (key, value, updated_at) VALUES ('theme', '\"dark\"', 0)",
                [],
            )
            .unwrap();

        // Detect the commit through the polling thread.
        match rx.recv_timeout(Duration::from_millis(3000)) {
            Ok(()) => {}
            Err(e) => panic!("expected a change signal within 3s, got {:?}", e),
        }
    }

    #[test]
    fn test_watch_changes_times_out() {
        let tmp = tempfile::NamedTempFile::new().unwrap();
        let path = tmp.path().to_path_buf();

        // Open the backend
        let backend = SqliteBackend::open(&path).unwrap();
        let rx = backend
            .watch_changes()
            .expect("polling thread should be running");

        // Detect the commit through the polling thread.
        match rx.recv_timeout(Duration::from_millis(1)) {
            Ok(()) => panic!("did not expect a signal"),
            Err(_e) => {}
        }
    }

    #[test]
    fn test_open_with_options_can_disable_watch_changes() {
        let tmp = tempfile::NamedTempFile::new().unwrap();
        let path = tmp.path().to_path_buf();
        let options = SqliteOptions {
            cross_process: false,
            ..SqliteOptions::default()
        };

        let backend = SqliteBackend::open_with_options(&path, options).unwrap();

        assert!(backend.watch_changes().is_none());
    }

    #[test]
    fn test_open_with_options_applies_pragmas() {
        let tmp = tempfile::NamedTempFile::new().unwrap();
        let path = tmp.path().to_path_buf();
        let options = SqliteOptions {
            cross_process: false,
            journal_mode: SqliteJournalMode::Delete,
            busy_timeout: Duration::from_millis(1234),
            synchronous: SqliteSynchronous::Full,
            ..SqliteOptions::default()
        };

        let backend = SqliteBackend::open_with_options(&path, options).unwrap();
        let conn = backend.conn.lock().unwrap();

        let journal_mode: String = conn
            .query_row("PRAGMA journal_mode", [], |row| row.get(0))
            .unwrap();
        let busy_timeout: i64 = conn
            .query_row("PRAGMA busy_timeout", [], |row| row.get(0))
            .unwrap();
        let synchronous: i64 = conn
            .query_row("PRAGMA synchronous", [], |row| row.get(0))
            .unwrap();

        assert_eq!(journal_mode, "delete");
        assert_eq!(busy_timeout, 1234);
        assert_eq!(synchronous, 2);
    }
}