ryu_storage/
lib.rs

1//! Plugin-owned key/value storage — the extracted `storage` primitive crate.
2//!
3//! Each plugin gets an isolated, namespaced KV space exposed **only** through the
4//! plugin-host `storage` capability (gated by the `storage:kv` grant). This is
5//! where a plugin keeps durable state instead of Core growing bespoke columns for
6//! it — e.g. the goal plugin's per-conversation completion condition + turn count
7//! live here (key = conversation id), not on the `conversations` table.
8//!
9//! Placement (Core vs Gateway): this stores *what a plugin is tracking* — it
10//! decides what runs, not what is allowed — so it is Core-tier. Rows are
11//! namespaced by `(plugin_id, namespace, key)` so one plugin can never read
12//! another's state.
13//!
14//! This crate is a **pure** primitive: [`PluginStorage::open`] takes an explicit
15//! db path, so the crate has ZERO dependency on `apps/core`. The single kernel
16//! coupling — choosing the default `~/.ryu/plugin-storage.db` path — and the
17//! process-global handle stay Core-side as wiring (`apps/core/src/plugin_storage`).
18
19use anyhow::{Context, Result};
20use rusqlite::{params, Connection, OptionalExtension};
21use std::path::PathBuf;
22use std::sync::Arc;
23use std::time::{SystemTime, UNIX_EPOCH};
24use tokio::sync::Mutex;
25
26fn now_millis() -> i64 {
27    SystemTime::now()
28        .duration_since(UNIX_EPOCH)
29        .map(|d| d.as_millis() as i64)
30        .unwrap_or(0)
31}
32
33/// SQLite-backed per-plugin KV store. Cheap to clone (wraps an `Arc`).
34#[derive(Clone)]
35pub struct PluginStorage {
36    conn: Arc<Mutex<Connection>>,
37}
38
39impl PluginStorage {
40    /// Open (or create) the store at a specific path and run migrations.
41    pub fn open(path: PathBuf) -> Result<Self> {
42        if let Some(parent) = path.parent() {
43            std::fs::create_dir_all(parent)
44                .with_context(|| format!("creating db dir {}", parent.display()))?;
45        }
46        let conn = Connection::open(&path)
47            .with_context(|| format!("opening plugin-storage db {}", path.display()))?;
48        Self::init_schema(&conn)?;
49        Ok(Self {
50            conn: Arc::new(Mutex::new(conn)),
51        })
52    }
53
54    /// In-memory store for tests.
55    pub fn in_memory() -> Result<Self> {
56        let conn = Connection::open_in_memory().context("opening in-memory plugin-storage db")?;
57        Self::init_schema(&conn)?;
58        Ok(Self {
59            conn: Arc::new(Mutex::new(conn)),
60        })
61    }
62
63    fn init_schema(conn: &Connection) -> Result<()> {
64        conn.execute_batch(
65            "PRAGMA journal_mode = WAL;
66             CREATE TABLE IF NOT EXISTS plugin_kv (
67                 plugin_id  TEXT NOT NULL,
68                 namespace  TEXT NOT NULL,
69                 key        TEXT NOT NULL,
70                 value      TEXT NOT NULL,
71                 updated_at INTEGER NOT NULL,
72                 PRIMARY KEY (plugin_id, namespace, key)
73             );",
74        )
75        .context("initializing plugin-storage schema")?;
76        Ok(())
77    }
78
79    /// Read a value. `Ok(None)` when the key is unset.
80    pub async fn get(&self, plugin_id: &str, namespace: &str, key: &str) -> Result<Option<String>> {
81        let conn = self.conn.lock().await;
82        let v = conn
83            .query_row(
84                "SELECT value FROM plugin_kv WHERE plugin_id = ?1 AND namespace = ?2 AND key = ?3",
85                params![plugin_id, namespace, key],
86                |row| row.get::<_, String>(0),
87            )
88            .optional()
89            .context("reading plugin_kv")?;
90        Ok(v)
91    }
92
93    /// Upsert a value.
94    pub async fn set(
95        &self,
96        plugin_id: &str,
97        namespace: &str,
98        key: &str,
99        value: &str,
100    ) -> Result<()> {
101        let conn = self.conn.lock().await;
102        conn.execute(
103            "INSERT INTO plugin_kv (plugin_id, namespace, key, value, updated_at)
104             VALUES (?1, ?2, ?3, ?4, ?5)
105             ON CONFLICT(plugin_id, namespace, key)
106             DO UPDATE SET value = excluded.value, updated_at = excluded.updated_at",
107            params![plugin_id, namespace, key, value, now_millis()],
108        )
109        .context("writing plugin_kv")?;
110        Ok(())
111    }
112
113    /// Delete a value (no-op if absent).
114    pub async fn delete(&self, plugin_id: &str, namespace: &str, key: &str) -> Result<()> {
115        let conn = self.conn.lock().await;
116        conn.execute(
117            "DELETE FROM plugin_kv WHERE plugin_id = ?1 AND namespace = ?2 AND key = ?3",
118            params![plugin_id, namespace, key],
119        )
120        .context("deleting plugin_kv")?;
121        Ok(())
122    }
123
124    /// List the keys a plugin has set within a namespace (newest first).
125    pub async fn keys(&self, plugin_id: &str, namespace: &str) -> Result<Vec<String>> {
126        let conn = self.conn.lock().await;
127        let mut stmt = conn
128            .prepare(
129                "SELECT key FROM plugin_kv WHERE plugin_id = ?1 AND namespace = ?2
130                 ORDER BY updated_at DESC",
131            )
132            .context("preparing plugin_kv keys query")?;
133        let rows = stmt
134            .query_map(params![plugin_id, namespace], |row| row.get::<_, String>(0))
135            .context("querying plugin_kv keys")?;
136        let mut out = Vec::new();
137        for r in rows {
138            out.push(r.context("reading plugin_kv key row")?);
139        }
140        Ok(out)
141    }
142}
143
144#[cfg(test)]
145mod tests {
146    use super::*;
147
148    #[tokio::test]
149    async fn set_get_delete_roundtrip() {
150        let s = PluginStorage::in_memory().unwrap();
151        assert_eq!(s.get("p", "ns", "k").await.unwrap(), None);
152        s.set("p", "ns", "k", "v1").await.unwrap();
153        assert_eq!(s.get("p", "ns", "k").await.unwrap().as_deref(), Some("v1"));
154        // Upsert overwrites.
155        s.set("p", "ns", "k", "v2").await.unwrap();
156        assert_eq!(s.get("p", "ns", "k").await.unwrap().as_deref(), Some("v2"));
157        s.delete("p", "ns", "k").await.unwrap();
158        assert_eq!(s.get("p", "ns", "k").await.unwrap(), None);
159    }
160
161    #[tokio::test]
162    async fn plugins_are_isolated_by_id_and_namespace() {
163        let s = PluginStorage::in_memory().unwrap();
164        s.set("plugin-a", "default", "shared", "a").await.unwrap();
165        s.set("plugin-b", "default", "shared", "b").await.unwrap();
166        // Same key, different plugin → isolated.
167        assert_eq!(
168            s.get("plugin-a", "default", "shared")
169                .await
170                .unwrap()
171                .as_deref(),
172            Some("a")
173        );
174        assert_eq!(
175            s.get("plugin-b", "default", "shared")
176                .await
177                .unwrap()
178                .as_deref(),
179            Some("b")
180        );
181        // Same plugin, different namespace → isolated.
182        s.set("plugin-a", "other", "shared", "a2").await.unwrap();
183        assert_eq!(
184            s.get("plugin-a", "default", "shared")
185                .await
186                .unwrap()
187                .as_deref(),
188            Some("a")
189        );
190    }
191
192    #[tokio::test]
193    async fn keys_lists_namespaced_keys() {
194        let s = PluginStorage::in_memory().unwrap();
195        s.set("p", "goals", "conv-1", "x").await.unwrap();
196        s.set("p", "goals", "conv-2", "y").await.unwrap();
197        s.set("p", "other", "conv-3", "z").await.unwrap();
198        let mut keys = s.keys("p", "goals").await.unwrap();
199        keys.sort();
200        assert_eq!(keys, vec!["conv-1".to_string(), "conv-2".to_string()]);
201    }
202}
ryu_storage/lib.rs

ryu_storage/
lib.rs
