do-memory-storage-redb 0.1.30

redb embedded storage backend for do-memory-core episodic learning system (cache layer)
Documentation
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
//! Redb cache layer for episode relationships.

use crate::{RELATIONSHIPS_TABLE, RedbStorage, Result};
#[allow(unused_imports)] // False positive - import is used in error mapping
use do_memory_core::Error;
use do_memory_core::episode::{Direction, EpisodeRelationship, RelationshipType};
use redb::{ReadableDatabase, ReadableTable, ReadableTableMetadata};
use tracing::debug;
use uuid::Uuid;

impl RedbStorage {
    // StorageBackend trait implementations

    /// Store a relationship (StorageBackend trait)
    pub async fn store_relationship(&self, relationship: &EpisodeRelationship) -> Result<()> {
        self.cache_relationship(relationship)
    }

    /// Remove a relationship (StorageBackend trait)
    pub async fn remove_relationship(&self, relationship_id: Uuid) -> Result<()> {
        self.remove_cached_relationship(relationship_id)
    }

    /// Get relationships (StorageBackend trait)
    pub async fn get_relationships(
        &self,
        episode_id: Uuid,
        direction: Direction,
    ) -> Result<Vec<EpisodeRelationship>> {
        self.get_cached_relationships(episode_id, direction)
    }

    /// Check if relationship exists (StorageBackend trait)
    pub async fn relationship_exists(
        &self,
        from_episode_id: Uuid,
        to_episode_id: Uuid,
        relationship_type: RelationshipType,
    ) -> Result<bool> {
        let relationships = self.get_cached_relationships(from_episode_id, Direction::Outgoing)?;
        Ok(relationships
            .iter()
            .any(|r| r.to_episode_id == to_episode_id && r.relationship_type == relationship_type))
    }

    // Original redb-specific methods

    /// Cache a relationship
    pub fn cache_relationship(&self, relationship: &EpisodeRelationship) -> Result<()> {
        let write_txn = self
            .db
            .begin_write()
            .map_err(|e| do_memory_core::Error::Storage(format!("Begin write failed: {}", e)))?;
        {
            let mut table = write_txn
                .open_table(RELATIONSHIPS_TABLE)
                .map_err(|e| do_memory_core::Error::Storage(format!("Open table failed: {}", e)))?;
            let key = relationship.id.to_string();
            let value = postcard::to_allocvec(relationship).map_err(|e| {
                do_memory_core::Error::Storage(format!("Serialization error: {}", e))
            })?;
            table
                .insert(key.as_str(), value.as_slice())
                .map_err(|e| do_memory_core::Error::Storage(format!("Insert failed: {}", e)))?;
        }
        write_txn
            .commit()
            .map_err(|e| do_memory_core::Error::Storage(format!("Commit failed: {}", e)))?;

        debug!("Cached relationship {} in redb", relationship.id);
        Ok(())
    }

    /// Get a cached relationship by ID
    pub fn get_cached_relationship(
        &self,
        relationship_id: Uuid,
    ) -> Result<Option<EpisodeRelationship>> {
        let read_txn = self
            .db
            .begin_read()
            .map_err(|e| do_memory_core::Error::Storage(format!("Begin read failed: {}", e)))?;
        let table = read_txn
            .open_table(RELATIONSHIPS_TABLE)
            .map_err(|e| do_memory_core::Error::Storage(format!("Open table failed: {}", e)))?;
        let key = relationship_id.to_string();

        match table
            .get(key.as_str())
            .map_err(|e| do_memory_core::Error::Storage(format!("Get failed: {}", e)))?
        {
            Some(value) => {
                let bytes = value.value();
                let relationship: EpisodeRelationship =
                    postcard::from_bytes(bytes).map_err(|e| {
                        do_memory_core::Error::Storage(format!("Deserialization error: {}", e))
                    })?;
                Ok(Some(relationship))
            }
            None => Ok(None),
        }
    }

    /// Remove a relationship from cache
    pub fn remove_cached_relationship(&self, relationship_id: Uuid) -> Result<()> {
        let write_txn = self
            .db
            .begin_write()
            .map_err(|e| do_memory_core::Error::Storage(format!("Begin write failed: {}", e)))?;
        {
            let mut table = write_txn
                .open_table(RELATIONSHIPS_TABLE)
                .map_err(|e| do_memory_core::Error::Storage(format!("Open table failed: {}", e)))?;
            let key = relationship_id.to_string();
            table
                .remove(key.as_str())
                .map_err(|e| do_memory_core::Error::Storage(format!("Remove failed: {}", e)))?;
        }
        write_txn
            .commit()
            .map_err(|e| do_memory_core::Error::Storage(format!("Commit failed: {}", e)))?;

        debug!("Removed relationship {} from cache", relationship_id);
        Ok(())
    }

    /// Get all cached relationships for an episode
    pub fn get_cached_relationships(
        &self,
        episode_id: Uuid,
        direction: Direction,
    ) -> Result<Vec<EpisodeRelationship>> {
        let read_txn = self
            .db
            .begin_read()
            .map_err(|e| do_memory_core::Error::Storage(format!("Begin read failed: {}", e)))?;
        let table = read_txn
            .open_table(RELATIONSHIPS_TABLE)
            .map_err(|e| do_memory_core::Error::Storage(format!("Open table failed: {}", e)))?;

        let mut relationships = Vec::new();
        let iter = table.iter().map_err(|e| {
            do_memory_core::Error::Storage(format!("Iterator creation failed: {}", e))
        })?;

        for item in iter {
            let (_, value) = item.map_err(|e| {
                do_memory_core::Error::Storage(format!("Iterator next failed: {}", e))
            })?;
            let bytes = value.value();
            let relationship: EpisodeRelationship = postcard::from_bytes(bytes).map_err(|e| {
                do_memory_core::Error::Storage(format!("Deserialization error: {}", e))
            })?;

            let matches = match direction {
                Direction::Outgoing => relationship.from_episode_id == episode_id,
                Direction::Incoming => relationship.to_episode_id == episode_id,
                Direction::Both => {
                    relationship.from_episode_id == episode_id
                        || relationship.to_episode_id == episode_id
                }
            };

            if matches {
                relationships.push(relationship);
            }
        }

        debug!(
            "Found {} cached relationships for episode {} (direction: {:?})",
            relationships.len(),
            episode_id,
            direction
        );

        Ok(relationships)
    }

    /// Clear all cached relationships
    pub fn clear_relationships_cache(&self) -> Result<()> {
        let write_txn = self
            .db
            .begin_write()
            .map_err(|e| do_memory_core::Error::Storage(format!("Begin write failed: {}", e)))?;
        {
            let mut table = write_txn
                .open_table(RELATIONSHIPS_TABLE)
                .map_err(|e| do_memory_core::Error::Storage(format!("Open table failed: {}", e)))?;
            // Remove all entries
            let keys: Vec<String> = {
                let iter = table.iter().map_err(|e| {
                    do_memory_core::Error::Storage(format!("Iterator creation failed: {}", e))
                })?;
                let mut keys = Vec::new();
                for item in iter {
                    let (key, _) = item.map_err(|e| {
                        do_memory_core::Error::Storage(format!("Iterator next failed: {}", e))
                    })?;
                    keys.push(key.value().to_string());
                }
                keys
            };

            for key in keys {
                table
                    .remove(key.as_str())
                    .map_err(|e| do_memory_core::Error::Storage(format!("Remove failed: {}", e)))?;
            }
        }
        write_txn
            .commit()
            .map_err(|e| do_memory_core::Error::Storage(format!("Commit failed: {}", e)))?;

        debug!("Cleared all cached relationships");
        Ok(())
    }

    /// Get count of cached relationships
    pub fn count_cached_relationships(&self) -> Result<usize> {
        let read_txn = self
            .db
            .begin_read()
            .map_err(|e| do_memory_core::Error::Storage(format!("Begin read failed: {}", e)))?;
        let table = read_txn
            .open_table(RELATIONSHIPS_TABLE)
            .map_err(|e| do_memory_core::Error::Storage(format!("Open table failed: {}", e)))?;
        let count = table.len().map_err(|e| {
            do_memory_core::Error::Storage(format!("Failed to get table length: {}", e))
        })? as usize;
        Ok(count)
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use do_memory_core::episode::RelationshipType;
    use tempfile::TempDir;

    async fn create_test_storage() -> (RedbStorage, TempDir) {
        let dir = TempDir::new().expect("Failed to create temp dir");
        let db_path = dir.path().join("test.redb");
        let storage = RedbStorage::new(&db_path)
            .await
            .expect("Failed to create storage");
        (storage, dir)
    }

    fn create_test_relationship(from_id: Uuid, to_id: Uuid) -> EpisodeRelationship {
        EpisodeRelationship::with_reason(
            from_id,
            to_id,
            RelationshipType::ParentChild,
            "Test relationship".to_string(),
        )
    }

    #[tokio::test]
    async fn test_cache_and_get_relationship() {
        let (storage, _dir) = create_test_storage().await;
        let from_id = Uuid::new_v4();
        let to_id = Uuid::new_v4();
        let relationship = create_test_relationship(from_id, to_id);
        let rel_id = relationship.id;

        // Cache the relationship
        storage
            .cache_relationship(&relationship)
            .expect("Failed to cache relationship");

        // Retrieve it
        let cached = storage
            .get_cached_relationship(rel_id)
            .expect("Failed to get relationship");
        assert!(cached.is_some());
        let cached_rel = cached.unwrap();
        assert_eq!(cached_rel.id, rel_id);
        assert_eq!(cached_rel.from_episode_id, from_id);
        assert_eq!(cached_rel.to_episode_id, to_id);
    }

    #[tokio::test]
    async fn test_remove_cached_relationship() {
        let (storage, _dir) = create_test_storage().await;
        let from_id = Uuid::new_v4();
        let to_id = Uuid::new_v4();
        let relationship = create_test_relationship(from_id, to_id);
        let rel_id = relationship.id;

        storage
            .cache_relationship(&relationship)
            .expect("Failed to cache relationship");

        // Remove it
        storage
            .remove_cached_relationship(rel_id)
            .expect("Failed to remove relationship");

        // Verify it's gone
        let cached = storage
            .get_cached_relationship(rel_id)
            .expect("Failed to get relationship");
        assert!(cached.is_none());
    }

    #[tokio::test]
    async fn test_get_cached_relationships_outgoing() {
        let (storage, _dir) = create_test_storage().await;
        let from_id = Uuid::new_v4();
        let to_id1 = Uuid::new_v4();
        let to_id2 = Uuid::new_v4();

        let rel1 = create_test_relationship(from_id, to_id1);
        let rel2 = create_test_relationship(from_id, to_id2);

        storage.cache_relationship(&rel1).expect("Failed to cache");
        storage.cache_relationship(&rel2).expect("Failed to cache");

        let relationships = storage
            .get_cached_relationships(from_id, Direction::Outgoing)
            .expect("Failed to get relationships");

        assert_eq!(relationships.len(), 2);
        assert!(relationships.iter().any(|r| r.to_episode_id == to_id1));
        assert!(relationships.iter().any(|r| r.to_episode_id == to_id2));
    }

    #[tokio::test]
    async fn test_get_cached_relationships_incoming() {
        let (storage, _dir) = create_test_storage().await;
        let to_id = Uuid::new_v4();
        let from_id1 = Uuid::new_v4();
        let from_id2 = Uuid::new_v4();

        let rel1 = create_test_relationship(from_id1, to_id);
        let rel2 = create_test_relationship(from_id2, to_id);

        storage.cache_relationship(&rel1).expect("Failed to cache");
        storage.cache_relationship(&rel2).expect("Failed to cache");

        let relationships = storage
            .get_cached_relationships(to_id, Direction::Incoming)
            .expect("Failed to get relationships");

        assert_eq!(relationships.len(), 2);
        assert!(relationships.iter().any(|r| r.from_episode_id == from_id1));
        assert!(relationships.iter().any(|r| r.from_episode_id == from_id2));
    }

    #[tokio::test]
    async fn test_clear_relationships_cache() {
        let (storage, _dir) = create_test_storage().await;
        let from_id = Uuid::new_v4();
        let to_id = Uuid::new_v4();

        for _ in 0..5 {
            let rel = create_test_relationship(from_id, to_id);
            storage.cache_relationship(&rel).expect("Failed to cache");
        }

        let count_before = storage
            .count_cached_relationships()
            .expect("Failed to count");
        assert_eq!(count_before, 5);

        storage
            .clear_relationships_cache()
            .expect("Failed to clear cache");

        let count_after = storage
            .count_cached_relationships()
            .expect("Failed to count");
        assert_eq!(count_after, 0);
    }

    #[tokio::test]
    async fn test_count_cached_relationships() {
        let (storage, _dir) = create_test_storage().await;

        // Add at least one relationship to ensure table exists
        let from_id = Uuid::new_v4();
        let to_id = Uuid::new_v4();
        let rel = create_test_relationship(from_id, to_id);
        storage.cache_relationship(&rel).expect("Failed to cache");

        let count_initial = storage
            .count_cached_relationships()
            .expect("Failed to count");
        assert_eq!(count_initial, 1);

        // Add 2 more relationships (total should be 3)
        for i in 0..2 {
            let from_id = Uuid::new_v4();
            let to_id = Uuid::new_v4();
            let rel = create_test_relationship(from_id, to_id);
            storage.cache_relationship(&rel).expect("Failed to cache");

            let count = storage
                .count_cached_relationships()
                .expect("Failed to count");
            assert_eq!(count, i + 2); // +2 because we start at 1
        }
    }
}