interlink/
store.rs

1//! A durable, keep-until-acked FIFO queue — one logical queue per recipient key.
2//!
3//! Backed by redb (pure Rust, ACID) on disk, or an in-memory backend for
4//! ephemeral use (tests, and the bus with no `--db`). Same code path either way.
5//!
6//! Values are opaque bytes, so the same store serves the bus (message
7//! envelopes) and the agent's outbound queue. Ordering is a global monotonic
8//! sequence, so a prefix range scan over a recipient yields FIFO. A message
9//! stays until [`Store::ack`]; redelivery after a crash is safe because the
10//! receiver dedupes by `msg_id`.
11//!
12//! redb's API is synchronous; each call runs on a blocking thread so the surface
13//! the rest of the async code sees is `async`. When Turso's pure-Rust SDK
14//! matures this module is the single seam to swap.
15
16use std::path::Path;
17use std::sync::Arc;
18
19use anyhow::{Context, Result};
20use redb::{Database, ReadableDatabase, ReadableTable, TableDefinition, TableError};
21use serde::{Deserialize, Serialize};
22
23const MESSAGES: TableDefinition<&str, &[u8]> = TableDefinition::new("messages");
24const META: TableDefinition<&str, u64> = TableDefinition::new("meta");
25const NEXT_SEQ: &str = "next_seq";
26
27// The message log lives in the same file as the queue but in its own tables.
28// LOG holds one record per msg_id (the current state); LOG_INDEX orders records
29// per peer by timestamp so a prefix scan yields a conversation in time order.
30const LOG: TableDefinition<&str, &[u8]> = TableDefinition::new("log");
31const LOG_INDEX: TableDefinition<&str, &str> = TableDefinition::new("log_index");
32
33#[derive(Debug, Clone, Copy, PartialEq, Serialize, Deserialize)]
34#[serde(rename_all = "snake_case")]
35pub enum Dir {
36    In,
37    Out,
38}
39
40/// One entry in the local conversation log. `text` may be `None` for records
41/// where only the fact of a message (not its body) needs keeping.
42#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
43pub struct LogRecord {
44    pub msg_id: String,
45    pub dir: Dir,
46    pub peer: String,
47    pub text: Option<String>,
48    pub ts: u64,
49    pub state: String,
50}
51
52fn log_index_key(peer: &str, ts: u64, msg_id: &str) -> String {
53    format!("{peer}\u{0}{ts:020}\u{0}{msg_id}")
54}
55
56#[derive(Clone)]
57pub struct Store {
58    db: Arc<Database>,
59}
60
61/// NUL separates the recipient from the zero-padded seq, so a range over
62/// `"{recipient}\0" .. "{recipient}\u{1}"` selects exactly that recipient's
63/// messages, in insertion order.
64fn msg_key(recipient: &str, seq: u64) -> String {
65    format!("{recipient}\u{0}{seq:020}")
66}
67
68fn recipient_bounds(recipient: &str) -> (String, String) {
69    (format!("{recipient}\u{0}"), format!("{recipient}\u{1}"))
70}
71
72impl Store {
73    /// On-disk durable store (created if absent).
74    pub fn on_disk(path: &Path) -> Result<Self> {
75        let db = Database::create(path)
76            .with_context(|| format!("opening store at {}", path.display()))?;
77        Ok(Self { db: Arc::new(db) })
78    }
79
80    /// Ephemeral in-memory store — same API, nothing persists.
81    pub fn in_memory() -> Result<Self> {
82        let db = Database::builder()
83            .create_with_backend(redb::backends::InMemoryBackend::new())
84            .context("creating in-memory store")?;
85        Ok(Self { db: Arc::new(db) })
86    }
87
88    /// Append `value` to `recipient`'s queue; returns the ack key.
89    pub async fn enqueue(&self, recipient: String, value: Vec<u8>) -> Result<String> {
90        let db = self.db.clone();
91        tokio::task::spawn_blocking(move || {
92            let wtx = db.begin_write()?;
93            let key;
94            {
95                let mut meta = wtx.open_table(META)?;
96                let seq = meta.get(NEXT_SEQ)?.map(|v| v.value()).unwrap_or(0);
97                meta.insert(NEXT_SEQ, seq + 1)?;
98                key = msg_key(&recipient, seq);
99                let mut msgs = wtx.open_table(MESSAGES)?;
100                msgs.insert(key.as_str(), value.as_slice())?;
101            }
102            wtx.commit()?;
103            Ok::<_, anyhow::Error>(key)
104        })
105        .await?
106    }
107
108    /// The oldest un-acked `(key, value)` for `recipient`, without removing it.
109    pub async fn peek_oldest(&self, recipient: String) -> Result<Option<(String, Vec<u8>)>> {
110        let db = self.db.clone();
111        tokio::task::spawn_blocking(move || {
112            let rtx = db.begin_read()?;
113            let msgs = match rtx.open_table(MESSAGES) {
114                Ok(t) => t,
115                Err(TableError::TableDoesNotExist(_)) => return Ok(None),
116                Err(e) => return Err(e.into()),
117            };
118            let (lo, hi) = recipient_bounds(&recipient);
119            match msgs.range(lo.as_str()..hi.as_str())?.next() {
120                Some(entry) => {
121                    let (k, v) = entry?;
122                    Ok(Some((k.value().to_string(), v.value().to_vec())))
123                }
124                None => Ok(None),
125            }
126        })
127        .await?
128    }
129
130    /// Remove an acked message by key. Idempotent (removing an absent key is ok).
131    pub async fn ack(&self, key: String) -> Result<()> {
132        let db = self.db.clone();
133        tokio::task::spawn_blocking(move || {
134            let wtx = db.begin_write()?;
135            {
136                let mut msgs = wtx.open_table(MESSAGES)?;
137                msgs.remove(key.as_str())?;
138            }
139            wtx.commit()?;
140            Ok::<_, anyhow::Error>(())
141        })
142        .await?
143    }
144
145    /// Count of un-acked messages for `recipient`.
146    pub async fn depth(&self, recipient: String) -> Result<usize> {
147        let db = self.db.clone();
148        tokio::task::spawn_blocking(move || {
149            let rtx = db.begin_read()?;
150            let msgs = match rtx.open_table(MESSAGES) {
151                Ok(t) => t,
152                Err(TableError::TableDoesNotExist(_)) => return Ok(0),
153                Err(e) => return Err(e.into()),
154            };
155            let (lo, hi) = recipient_bounds(&recipient);
156            Ok::<_, anyhow::Error>(msgs.range(lo.as_str()..hi.as_str())?.count())
157        })
158        .await?
159    }
160
161    /// Every un-acked `(key, value)` for `recipient`, in FIFO order. Used to
162    /// inspect a queue (e.g. list what's still pending in the agent's outbox).
163    pub async fn list(&self, recipient: String) -> Result<Vec<(String, Vec<u8>)>> {
164        let db = self.db.clone();
165        tokio::task::spawn_blocking(move || {
166            let rtx = db.begin_read()?;
167            let msgs = match rtx.open_table(MESSAGES) {
168                Ok(t) => t,
169                Err(TableError::TableDoesNotExist(_)) => return Ok(Vec::new()),
170                Err(e) => return Err(e.into()),
171            };
172            let (lo, hi) = recipient_bounds(&recipient);
173            let mut out = Vec::new();
174            for entry in msgs.range(lo.as_str()..hi.as_str())? {
175                let (k, v) = entry?;
176                out.push((k.value().to_string(), v.value().to_vec()));
177            }
178            Ok::<_, anyhow::Error>(out)
179        })
180        .await?
181    }
182
183    /// Insert or overwrite a log record (keyed by msg_id) and its per-peer index.
184    pub async fn log_put(&self, rec: LogRecord) -> Result<()> {
185        let db = self.db.clone();
186        tokio::task::spawn_blocking(move || {
187            let idx_key = log_index_key(&rec.peer, rec.ts, &rec.msg_id);
188            let bytes = serde_json::to_vec(&rec)?;
189            let wtx = db.begin_write()?;
190            {
191                let mut log = wtx.open_table(LOG)?;
192                log.insert(rec.msg_id.as_str(), bytes.as_slice())?;
193                let mut idx = wtx.open_table(LOG_INDEX)?;
194                idx.insert(idx_key.as_str(), rec.msg_id.as_str())?;
195            }
196            wtx.commit()?;
197            Ok::<_, anyhow::Error>(())
198        })
199        .await?
200    }
201
202    /// Update the `state` of a logged message. No-op if the msg_id is unknown.
203    pub async fn log_set_state(&self, msg_id: String, state: String) -> Result<()> {
204        let db = self.db.clone();
205        tokio::task::spawn_blocking(move || {
206            let wtx = db.begin_write()?;
207            {
208                let mut log = wtx.open_table(LOG)?;
209                let current = log.get(msg_id.as_str())?.map(|g| g.value().to_vec());
210                if let Some(cur) = current {
211                    let mut rec: LogRecord = serde_json::from_slice(&cur)?;
212                    rec.state = state;
213                    let bytes = serde_json::to_vec(&rec)?;
214                    log.insert(msg_id.as_str(), bytes.as_slice())?;
215                }
216            }
217            wtx.commit()?;
218            Ok::<_, anyhow::Error>(())
219        })
220        .await?
221    }
222
223    /// The log record for a single msg_id, if any.
224    pub async fn log_get(&self, msg_id: String) -> Result<Option<LogRecord>> {
225        let db = self.db.clone();
226        tokio::task::spawn_blocking(move || {
227            let rtx = db.begin_read()?;
228            let log = match rtx.open_table(LOG) {
229                Ok(t) => t,
230                Err(TableError::TableDoesNotExist(_)) => return Ok(None),
231                Err(e) => return Err(e.into()),
232            };
233            match log.get(msg_id.as_str())? {
234                Some(g) => Ok(Some(serde_json::from_slice(g.value())?)),
235                None => Ok(None),
236            }
237        })
238        .await?
239    }
240
241    /// The most recent `limit` log records for `peer`, in chronological order.
242    pub async fn log_by_peer(&self, peer: String, limit: usize) -> Result<Vec<LogRecord>> {
243        let db = self.db.clone();
244        tokio::task::spawn_blocking(move || {
245            let rtx = db.begin_read()?;
246            let idx = match rtx.open_table(LOG_INDEX) {
247                Ok(t) => t,
248                Err(TableError::TableDoesNotExist(_)) => return Ok(Vec::new()),
249                Err(e) => return Err(e.into()),
250            };
251            let log = rtx.open_table(LOG)?;
252            let (lo, hi) = recipient_bounds(&peer);
253            let mut ids: Vec<String> = Vec::new();
254            for entry in idx.range(lo.as_str()..hi.as_str())? {
255                let (_k, v) = entry?;
256                ids.push(v.value().to_string());
257            }
258            if ids.len() > limit {
259                ids = ids.split_off(ids.len() - limit);
260            }
261            let mut out = Vec::with_capacity(ids.len());
262            for id in ids {
263                if let Some(g) = log.get(id.as_str())? {
264                    out.push(serde_json::from_slice(g.value())?);
265                }
266            }
267            Ok::<_, anyhow::Error>(out)
268        })
269        .await?
270    }
271}
272
273#[cfg(test)]
274mod tests {
275    use super::*;
276
277    fn b(s: &str) -> Vec<u8> {
278        s.as_bytes().to_vec()
279    }
280
281    #[tokio::test]
282    async fn enqueue_peek_ack_roundtrip() {
283        let s = Store::in_memory().unwrap();
284        let key = s.enqueue("alice".into(), b("hi")).await.unwrap();
285        let (k, v) = s.peek_oldest("alice".into()).await.unwrap().unwrap();
286        assert_eq!(k, key);
287        assert_eq!(v, b("hi"));
288        // peek does not remove
289        assert!(s.peek_oldest("alice".into()).await.unwrap().is_some());
290        s.ack(key).await.unwrap();
291        assert!(s.peek_oldest("alice".into()).await.unwrap().is_none());
292    }
293
294    #[tokio::test]
295    async fn fifo_order_within_recipient() {
296        let s = Store::in_memory().unwrap();
297        for m in ["one", "two", "three"] {
298            s.enqueue("bob".into(), b(m)).await.unwrap();
299        }
300        for expected in ["one", "two", "three"] {
301            let (k, v) = s.peek_oldest("bob".into()).await.unwrap().unwrap();
302            assert_eq!(v, b(expected));
303            s.ack(k).await.unwrap();
304        }
305        assert!(s.peek_oldest("bob".into()).await.unwrap().is_none());
306    }
307
308    #[tokio::test]
309    async fn recipients_are_isolated() {
310        let s = Store::in_memory().unwrap();
311        s.enqueue("alice".into(), b("for-alice")).await.unwrap();
312        s.enqueue("bob".into(), b("for-bob")).await.unwrap();
313        assert_eq!(s.depth("alice".into()).await.unwrap(), 1);
314        assert_eq!(s.depth("bob".into()).await.unwrap(), 1);
315        assert_eq!(
316            s.peek_oldest("bob".into()).await.unwrap().unwrap().1,
317            b("for-bob")
318        );
319    }
320
321    #[tokio::test]
322    async fn unacked_message_survives_reopen() {
323        // The whole point: a message persists across a bus restart until acked.
324        let dir = tempfile::tempdir().unwrap();
325        let path = dir.path().join("q.redb");
326        let key = {
327            let s = Store::on_disk(&path).unwrap();
328            s.enqueue("alice".into(), b("durable")).await.unwrap()
329        }; // Store (and its Database) dropped — simulates a restart
330        let s2 = Store::on_disk(&path).unwrap();
331        let (k, v) = s2.peek_oldest("alice".into()).await.unwrap().unwrap();
332        assert_eq!(k, key);
333        assert_eq!(v, b("durable"));
334        // and once acked, it's gone across another reopen
335        s2.ack(k).await.unwrap();
336        drop(s2);
337        let s3 = Store::on_disk(&path).unwrap();
338        assert!(s3.peek_oldest("alice".into()).await.unwrap().is_none());
339    }
340
341    #[tokio::test]
342    async fn ack_is_idempotent() {
343        let s = Store::in_memory().unwrap();
344        let key = s.enqueue("alice".into(), b("x")).await.unwrap();
345        s.ack(key.clone()).await.unwrap();
346        s.ack(key).await.unwrap(); // no error the second time
347        s.ack("never-existed".into()).await.unwrap();
348    }
349
350    fn rec(msg_id: &str, dir: Dir, peer: &str, text: Option<&str>, ts: u64) -> LogRecord {
351        LogRecord {
352            msg_id: msg_id.into(),
353            dir,
354            peer: peer.into(),
355            text: text.map(str::to_string),
356            ts,
357            state: "pending".into(),
358        }
359    }
360
361    #[tokio::test]
362    async fn log_put_get_and_set_state() {
363        let s = Store::in_memory().unwrap();
364        s.log_put(rec("m1", Dir::Out, "bob", Some("hi bob"), 10))
365            .await
366            .unwrap();
367        let got = s.log_get("m1".into()).await.unwrap().unwrap();
368        assert_eq!(got.peer, "bob");
369        assert_eq!(got.state, "pending");
370        s.log_set_state("m1".into(), "sent".into()).await.unwrap();
371        assert_eq!(s.log_get("m1".into()).await.unwrap().unwrap().state, "sent");
372        // unknown id is a no-op, not an error
373        s.log_set_state("nope".into(), "sent".into()).await.unwrap();
374        assert!(s.log_get("nope".into()).await.unwrap().is_none());
375    }
376
377    #[tokio::test]
378    async fn log_by_peer_is_chronological_and_limited() {
379        let s = Store::in_memory().unwrap();
380        s.log_put(rec("a", Dir::Out, "bob", Some("1"), 1))
381            .await
382            .unwrap();
383        s.log_put(rec("b", Dir::In, "bob", Some("2"), 2))
384            .await
385            .unwrap();
386        s.log_put(rec("c", Dir::Out, "bob", Some("3"), 3))
387            .await
388            .unwrap();
389        s.log_put(rec("z", Dir::In, "carol", Some("other"), 5))
390            .await
391            .unwrap();
392        let hist = s.log_by_peer("bob".into(), 2).await.unwrap();
393        assert_eq!(
394            hist.iter().map(|r| r.msg_id.as_str()).collect::<Vec<_>>(),
395            vec!["b", "c"],
396            "newest 2, in time order"
397        );
398        // carol's history is isolated from bob's
399        assert_eq!(s.log_by_peer("carol".into(), 10).await.unwrap().len(), 1);
400    }
401
402    #[tokio::test]
403    async fn scoped_body_is_not_persisted() {
404        let s = Store::in_memory().unwrap();
405        s.log_put(rec("s1", Dir::In, "carol", None, 7))
406            .await
407            .unwrap();
408        let got = s.log_get("s1".into()).await.unwrap().unwrap();
409        assert!(got.text.is_none(), "withheld body must stay withheld");
410    }
411
412    #[tokio::test]
413    async fn list_returns_all_queued_for_recipient() {
414        let s = Store::in_memory().unwrap();
415        s.enqueue("outbox".into(), b("j1")).await.unwrap();
416        s.enqueue("outbox".into(), b("j2")).await.unwrap();
417        let all = s.list("outbox".into()).await.unwrap();
418        assert_eq!(all.len(), 2);
419        assert_eq!(all[0].1, b("j1"));
420        assert_eq!(all[1].1, b("j2"));
421    }
422}
interlink/store.rs

interlink/
store.rs
