agileplus_sqlite/

storage.rs

//! SQLite-backed storage for intent graphs.

use std::path::Path;
use std::sync::{Arc, Mutex};

use agileplus_domain::intent_graph::IntentGraph;
use anyhow::{Context, Result};
use regex::Regex;
use rusqlite::functions::FunctionFlags;
use rusqlite::{params, Connection, Error as SqliteError, OptionalExtension};
use serde::{Deserialize, Serialize};

/// Thread-safe wrapper around a `rusqlite::Connection`.
#[derive(Clone)]
pub struct Storage {
    conn: Arc<Mutex<Connection>>,
}

impl std::fmt::Debug for Storage {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("Storage").finish_non_exhaustive()
    }
}

/// A free-form note attached to a graph.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct Note {
    pub id: i64,
    pub graph_id: String,
    pub body: String,
    pub created_at: String,
    pub updated_at: String,
}

impl Storage {
    /// Open a connection at `path` and apply all migrations.
    pub fn open(path: impl AsRef<Path>) -> Result<Self> {
        let conn = Connection::open(path.as_ref())
            .with_context(|| format!("opening db at {}", path.as_ref().display()))?;
        conn.execute_batch(
            "PRAGMA journal_mode = WAL;\n\
             PRAGMA synchronous = NORMAL;\n\
             PRAGMA foreign_keys = ON;",
        )
        .context("setting PRAGMAs on persistent connection")?;
        register_regexp(&conn).context("registering REGEXP SQL function")?;
        let storage = Self {
            conn: Arc::new(Mutex::new(conn)),
        };
        storage.migrate()?;
        Ok(storage)
    }

    /// Open an in-memory database. Useful for tests.
    pub fn in_memory() -> Result<Self> {
        let conn = Connection::open_in_memory().context("opening in-memory db")?;
        conn.execute_batch(
            "PRAGMA journal_mode = MEMORY;\n\
             PRAGMA synchronous = OFF;\n\
             PRAGMA foreign_keys = ON;",
        )
        .context("setting PRAGMAs on in-memory connection")?;
        register_regexp(&conn).context("registering REGEXP SQL function")?;
        let storage = Self {
            conn: Arc::new(Mutex::new(conn)),
        };
        storage.migrate()?;
        Ok(storage)
    }

    /// Apply all pending migrations to the open connection.
    pub fn migrate(&self) -> Result<()> {
        let conn = self
            .conn
            .lock()
            .map_err(|e| anyhow::anyhow!("storage mutex poisoned: {e}"))?;
        apply_migrations_to(&conn)
    }

    /// Persist a graph by the given `id`, replacing any previous version.
    pub fn save_graph(&self, id: &str, graph: &IntentGraph) -> Result<()> {
        let conn = self
            .conn
            .lock()
            .map_err(|e| anyhow::anyhow!("storage mutex poisoned: {e}"))?;
        let json = serde_json::to_string(graph).context("serializing intent graph")?;
        conn.execute(
            "INSERT INTO intent_graphs (id, payload, updated_at) \
             VALUES (?1, ?2, datetime('now')) \
             ON CONFLICT(id) DO UPDATE SET \
                payload = excluded.payload, \
                updated_at = excluded.updated_at",
            params![id, json],
        )
        .context("upserting intent_graphs row")?;
        Ok(())
    }

    /// Fetch a graph by id.
    pub fn load_graph(&self, id: &str) -> Result<Option<IntentGraph>> {
        let conn = self
            .conn
            .lock()
            .map_err(|e| anyhow::anyhow!("storage mutex poisoned: {e}"))?;
        let mut stmt = conn
            .prepare("SELECT payload FROM intent_graphs WHERE id = ?1")
            .context("preparing load_graph statement")?;
        let mut rows = stmt
            .query(params![id])
            .context("querying intent_graphs")?;
        let Some(row) = rows.next().context("iterating intent_graphs rows")? else {
            return Ok(None);
        };
        let payload: String = row.get(0).context("reading payload column")?;
        let graph: IntentGraph = serde_json::from_str(&payload)
            .with_context(|| format!("parsing graph payload for id={}", id))?;
        Ok(Some(graph))
    }

    /// List all stored graph ids, most-recently-updated first.
    pub fn list_graphs(&self) -> Result<Vec<String>> {
        let conn = self
            .conn
            .lock()
            .map_err(|e| anyhow::anyhow!("storage mutex poisoned: {e}"))?;
        let mut stmt = conn
            .prepare("SELECT id FROM intent_graphs ORDER BY updated_at DESC")
            .context("preparing list_graphs statement")?;
        let rows = stmt
            .query_map([], |r| r.get::<_, String>(0))
            .context("mapping list_graphs rows")?;
        let ids: Vec<String> = rows
            .collect::<rusqlite::Result<Vec<_>>>()
            .context("collecting list_graphs rows")?;
        Ok(ids)
    }

    /// Delete a graph by id; returns true if a row was removed.
    pub fn delete_graph(&self, id: &str) -> Result<bool> {
        let conn = self
            .conn
            .lock()
            .map_err(|e| anyhow::anyhow!("storage mutex poisoned: {e}"))?;
        let affected = conn
            .execute("DELETE FROM intent_graphs WHERE id = ?1", params![id])
            .context("deleting intent_graphs row")?;
        Ok(affected > 0)
    }

    /// Attach a tag to a graph. Returns `true` if the tag was newly inserted,
    /// `false` if the `(graph_id, tag)` pair was already present.
    pub fn add_tag(&self, graph_id: &str, tag: &str) -> Result<bool> {
        let conn = self
            .conn
            .lock()
            .map_err(|e| anyhow::anyhow!("storage mutex poisoned: {e}"))?;
        let affected = conn
            .execute(
                "INSERT OR IGNORE INTO graph_tags (graph_id, tag) VALUES (?1, ?2)",
                params![graph_id, tag],
            )
            .context("inserting graph_tags row")?;
        Ok(affected > 0)
    }

    /// Remove a tag from a graph. Returns `true` if a row was deleted,
    /// `false` if the tag was not present.
    pub fn remove_tag(&self, graph_id: &str, tag: &str) -> Result<bool> {
        let conn = self
            .conn
            .lock()
            .map_err(|e| anyhow::anyhow!("storage mutex poisoned: {e}"))?;
        let affected = conn
            .execute(
                "DELETE FROM graph_tags WHERE graph_id = ?1 AND tag = ?2",
                params![graph_id, tag],
            )
            .context("deleting graph_tags row")?;
        Ok(affected > 0)
    }

    /// List the tags attached to a graph, in alphabetical order.
    pub fn list_tags(&self, graph_id: &str) -> Result<Vec<String>> {
        let conn = self
            .conn
            .lock()
            .map_err(|e| anyhow::anyhow!("storage mutex poisoned: {e}"))?;
        let mut stmt = conn
            .prepare("SELECT tag FROM graph_tags WHERE graph_id = ?1 ORDER BY tag ASC")
            .context("preparing list_tags statement")?;
        let rows = stmt
            .query_map(params![graph_id], |r| r.get::<_, String>(0))
            .context("mapping list_tags rows")?;
        let tags: Vec<String> = rows
            .collect::<rusqlite::Result<Vec<_>>>()
            .context("collecting list_tags rows")?;
        Ok(tags)
    }

    /// Append a free-form note to a graph. Returns the new note id.
    pub fn add_note(&self, graph_id: &str, body: &str) -> Result<i64> {
        let conn = self
            .conn
            .lock()
            .map_err(|e| anyhow::anyhow!("storage mutex poisoned: {e}"))?;
        conn.execute(
            "INSERT INTO graph_notes (graph_id, body) VALUES (?1, ?2)",
            params![graph_id, body],
        )
        .context("inserting graph_notes row")?;
        Ok(conn.last_insert_rowid())
    }

    /// List notes attached to a graph, oldest first.
    pub fn list_notes(&self, graph_id: &str) -> Result<Vec<Note>> {
        let conn = self
            .conn
            .lock()
            .map_err(|e| anyhow::anyhow!("storage mutex poisoned: {e}"))?;
        let mut stmt = conn
            .prepare(
                "SELECT id, graph_id, body, created_at, updated_at \
                 FROM graph_notes WHERE graph_id = ?1 \
                 ORDER BY created_at ASC, id ASC",
            )
            .context("preparing list_notes statement")?;
        let rows = stmt
            .query_map(params![graph_id], |r| {
                Ok(Note {
                    id: r.get(0)?,
                    graph_id: r.get(1)?,
                    body: r.get(2)?,
                    created_at: r.get(3)?,
                    updated_at: r.get(4)?,
                })
            })
            .context("mapping list_notes rows")?;
        let notes: Vec<Note> = rows
            .collect::<rusqlite::Result<Vec<_>>>()
            .context("collecting list_notes rows")?;
        Ok(notes)
    }

    /// Delete a note by id, scoped to a specific graph. Returns `true` if a
    /// row was deleted, `false` otherwise (note did not exist for that graph).
    pub fn delete_note(&self, graph_id: &str, note_id: i64) -> Result<bool> {
        let conn = self
            .conn
            .lock()
            .map_err(|e| anyhow::anyhow!("storage mutex poisoned: {e}"))?;
        let affected = conn
            .execute(
                "DELETE FROM graph_notes WHERE graph_id = ?1 AND id = ?2",
                params![graph_id, note_id],
            )
            .context("deleting graph_notes row")?;
        Ok(affected > 0)
    }
}

/// Apply every bundled migration that has not already been recorded in
/// `migration_history`. Runs the whole batch inside a single transaction so
/// partially-applied migrations cannot leave the schema in an inconsistent
/// state.
fn apply_migrations_to(conn: &Connection) -> Result<()> {
    conn.execute_batch(
        "CREATE TABLE IF NOT EXISTS migration_history (\
            id TEXT PRIMARY KEY, \
            applied_at TEXT NOT NULL\
         )",
    )
    .context("creating migration_history table")?;

    let tx = conn
        .unchecked_transaction()
        .context("opening migration transaction")?;

    for (id, sql) in super::migrations::ALL {
        let already_applied: bool = tx
            .query_row(
                "SELECT 1 FROM migration_history WHERE id = ?1",
                params![id],
                |_| Ok(true),
            )
            .optional()
            .context("checking migration_history")?
            .unwrap_or(false);
        if already_applied {
            continue;
        }

        tx.execute_batch(sql)
            .with_context(|| format!("applying migration {id}"))?;
        tx.execute(
            "INSERT INTO migration_history (id, applied_at) VALUES (?1, datetime('now'))",
            params![id],
        )
        .with_context(|| format!("recording migration {id} in migration_history"))?;
    }

    tx.commit().context("committing migration transaction")?;
    Ok(())
}

/// Register a SQLite `REGEXP(pattern, text)` function backed by the
/// `regex` crate. Required because migration `025_create_intent_graph`
/// declares `CHECK (id REGEXP '...')` constraints, and SQLite ships
/// without a built-in `REGEXP` implementation.
fn register_regexp(conn: &Connection) -> rusqlite::Result<()> {
    conn.create_scalar_function(
        "REGEXP",
        2,
        FunctionFlags::SQLITE_UTF8 | FunctionFlags::SQLITE_DETERMINISTIC,
        |ctx| {
            let pattern: String = ctx.get(0)?;
            let text: String = ctx.get(1)?;
            match Regex::new(&pattern) {
                Ok(re) => Ok(re.is_match(&text)),
                Err(e) => Err(SqliteError::UserFunctionError(format!(
                    "invalid REGEXP pattern {pattern:?}: {e}"
                )
                .into())),
            }
        },
    )
}