guardian-db 0.15.0

High-performance, local-first decentralized database built on Rust and Iroh
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
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
use crate::address::Address;
use crate::data_store::Datastore;
use crate::guardian::error::{GuardianError, Result};
use sled::{Config, Db};
use std::collections::HashMap;
use std::path::{Path, PathBuf};
use std::sync::{Arc, Mutex};
use tracing::{Span, debug, error, info, instrument, warn};

#[allow(clippy::module_inception)]
pub mod level_down;
pub use level_down::LevelDownCache;

// Type aliases para simplificar tipos complexos
type DatastoreBox = Box<dyn Datastore + Send + Sync>;
type CleanupFn = Box<dyn FnOnce() -> Result<()> + Send + Sync>;
type NewCacheResult = Result<(DatastoreBox, CleanupFn)>;

/// Define as opções para a criação de um cache.
#[derive(Debug, Clone)]
pub struct Options {
    /// Span para logging estruturado com tracing.
    pub span: Option<Span>,
    /// Tamanho máximo do cache em bytes (padrão: 100MB)
    pub max_cache_size: Option<u64>,
    /// Modo de cache: persistente ou em memória
    pub cache_mode: CacheMode,
}

/// Modo de operação do cache
#[derive(Debug, Clone, PartialEq)]
pub enum CacheMode {
    /// Cache persistente no disco
    Persistent,
    /// Cache em memória (temporário)
    InMemory,
    /// Automático: detecta baseado no diretório
    Auto,
}

impl Default for Options {
    fn default() -> Self {
        Self {
            span: None,
            max_cache_size: Some(100 * 1024 * 1024), // 100MB
            cache_mode: CacheMode::Auto,
        }
    }
}

/// A trait `Cache` define a interface para um mecanismo de cache
/// para os bancos de dados GuardianDB.
pub trait Cache: Send + Sync {
    /// Cria uma nova instância de cache no caminho especificado
    /// Retorna um Datastore e uma função de cleanup
    #[allow(clippy::new_ret_no_self)]
    fn new(path: &str, opts: Option<Options>) -> NewCacheResult
    where
        Self: Sized,
    {
        SledCache::create_cache_instance(path, opts.unwrap_or_default())
    }

    /// Carrega um cache para um determinado endereço de banco de dados e um diretório raiz.
    fn load(&self, directory: &str, db_address: &dyn Address) -> Result<DatastoreBox>;

    /// Fecha um cache e todos os seus armazenamentos de dados associados.
    fn close(&mut self) -> Result<()>;

    /// Remove todos os dados em cache de um banco de dados.
    fn destroy(&self, directory: &str, db_address: &dyn Address) -> Result<()>;
}

/// Implementação de cache usando Sled como backend
pub struct SledCache {
    caches: Arc<Mutex<HashMap<String, Arc<SledDatastore>>>>,
    options: Options,
}

impl SledCache {
    /// Cria uma nova instância do SledCache
    pub fn new(opts: Options) -> Self {
        Self {
            caches: Arc::new(Mutex::new(HashMap::new())),
            options: opts,
        }
    }

    /// Factory method para criar instâncias de cache
    #[instrument(level = "info")]
    pub fn create_cache_instance(path: &str, opts: Options) -> NewCacheResult {
        info!("Creating cache instance: path={}", path);

        let datastore = SledDatastore::new(path, opts.clone())?;
        let path_clone = path.to_string();

        // Função de cleanup que remove o cache do disco (apenas se não for em memória)
        let cleanup: Box<dyn FnOnce() -> Result<()> + Send + Sync> = Box::new(move || {
            if path_clone != ":memory:" && Path::new(&path_clone).exists() {
                match std::fs::remove_dir_all(&path_clone) {
                    Ok(_) => {
                        debug!("Cache directory cleaned up: path={}", &path_clone);
                        Ok(())
                    }
                    Err(e) => {
                        warn!(
                            "Failed to cleanup cache directory: path={}, error={}",
                            &path_clone, e
                        );
                        Err(GuardianError::Other(format!(
                            "Failed to cleanup cache: {}",
                            e
                        )))
                    }
                }
            } else {
                Ok(())
            }
        });

        Ok((Box::new(datastore), cleanup))
    }

    /// Gera uma chave única para o cache baseada no diretório e endereço
    fn generate_cache_key(directory: &str, db_address: &dyn Address) -> String {
        let db_path = PathBuf::from(db_address.get_root().to_string()).join(db_address.get_path());
        PathBuf::from(directory)
            .join(db_path)
            .to_string_lossy()
            .to_string()
    }
}

impl Cache for SledCache {
    #[instrument(level = "info", skip(self, db_address))]
    fn load(
        &self,
        directory: &str,
        db_address: &dyn Address,
    ) -> Result<Box<dyn Datastore + Send + Sync>> {
        let cache_key = Self::generate_cache_key(directory, db_address);

        info!(
            "Loading cache: directory={}, cache_key={}",
            directory, &cache_key
        );

        let mut caches = self.caches.lock().unwrap();

        if let Some(existing_cache) = caches.get(&cache_key) {
            debug!("Using existing cache: cache_key={}", &cache_key);
            return Ok(Box::new(existing_cache.as_ref().clone()));
        }

        // Cria um novo cache se não existir
        let datastore = SledDatastore::new(&cache_key, self.options.clone())?;
        let arc_datastore = Arc::new(datastore.clone());
        caches.insert(cache_key.clone(), arc_datastore);

        info!("Created new cache: cache_key={}", &cache_key);
        Ok(Box::new(datastore))
    }

    #[instrument(level = "info", skip(self))]
    fn close(&mut self) -> Result<()> {
        info!("Closing all caches");

        let caches = {
            let mut cache_map = self.caches.lock().unwrap();
            let caches: Vec<Arc<SledDatastore>> = cache_map.values().cloned().collect();
            cache_map.clear();
            caches
        };

        for cache in caches {
            if let Err(e) = cache.close() {
                warn!("Failed to close cache: error={}", e);
            }
        }

        info!("All caches closed");
        Ok(())
    }

    #[instrument(level = "info", skip(self, db_address))]
    fn destroy(&self, directory: &str, db_address: &dyn Address) -> Result<()> {
        let cache_key = Self::generate_cache_key(directory, db_address);

        info!(
            "Destroying cache: directory={}, cache_key={}",
            directory, &cache_key
        );

        // Remove do mapa de caches
        let cache_to_close = {
            let mut caches = self.caches.lock().unwrap();
            caches.remove(&cache_key)
        };

        // Fecha o cache se existir
        if let Some(cache) = cache_to_close {
            cache.close()?;
        }

        // Remove arquivos do disco (apenas se não for em memória)
        if directory != ":memory:" && Path::new(&cache_key).exists() {
            std::fs::remove_dir_all(&cache_key).map_err(|e| {
                GuardianError::Other(format!("Failed to remove cache directory: {}", e))
            })?;

            info!("Cache directory removed: path={}", &cache_key);
        }

        Ok(())
    }
}

/// Implementação de Datastore usando Sled
#[derive(Clone)]
pub struct SledDatastore {
    db: Db,
    path: String,
    span: Span,
}

impl SledDatastore {
    /// Cria uma nova instância do SledDatastore
    #[instrument(level = "debug")]
    pub fn new(path: &str, opts: Options) -> Result<Self> {
        debug!("Creating SledDatastore: path={}", path);

        let db = if path == ":memory:" || matches!(opts.cache_mode, CacheMode::InMemory) {
            // Cache em memória
            debug!("Creating in-memory cache");
            Config::new().temporary(true).open().map_err(|e| {
                GuardianError::Store(format!("Failed to create in-memory cache: {}", e))
            })?
        } else {
            // Cache persistente
            debug!("Creating persistent cache: path={}", path);

            // Cria o diretório se não existir
            if let Some(parent) = Path::new(path).parent() {
                std::fs::create_dir_all(parent).map_err(|e| {
                    GuardianError::Store(format!("Failed to create cache directory: {}", e))
                })?;
            }

            let mut config = Config::new();

            // Configura tamanho máximo se especificado
            if let Some(max_size) = opts.max_cache_size {
                config = config.cache_capacity(max_size);
            }

            config.path(path).open().map_err(|e| {
                GuardianError::Store(format!("Failed to open cache at {}: {}", path, e))
            })?
        };

        info!(
            "SledDatastore created successfully: path={}, memory_mode={}",
            path,
            path == ":memory:"
        );

        Ok(Self {
            db,
            path: path.to_string(),
            span: opts.span.unwrap_or_else(tracing::Span::current),
        })
    }

    /// Retorna uma referência ao span de tracing para instrumentação
    pub fn span(&self) -> &Span {
        &self.span
    }

    /// Fecha o datastore
    #[instrument(level = "debug", skip(self))]
    pub fn close(&self) -> Result<()> {
        let _entered = self.span.enter();
        debug!("Closing SledDatastore: path={}", &self.path);

        self.db
            .flush()
            .map_err(|e| GuardianError::Store(format!("Failed to flush cache: {}", e)))?;

        info!("SledDatastore closed: path={}", &self.path);
        Ok(())
    }
}

#[async_trait::async_trait]
impl Datastore for SledDatastore {
    #[instrument(level = "debug", skip(self, key))]
    async fn get(&self, key: &[u8]) -> Result<Option<Vec<u8>>> {
        let _entered = self.span.enter();
        match self.db.get(key) {
            Ok(Some(value)) => {
                debug!("Cache hit: key_len={}", key.len());
                Ok(Some(value.to_vec()))
            }
            Ok(None) => {
                debug!("Cache miss: key_len={}", key.len());
                Ok(None)
            }
            Err(e) => {
                error!("Cache get error: key_len={}, error={}", key.len(), e);
                Err(GuardianError::Store(format!("Cache get error: {}", e)))
            }
        }
    }

    #[instrument(level = "debug", skip(self, key, value))]
    async fn put(&self, key: &[u8], value: &[u8]) -> Result<()> {
        let _entered = self.span.enter();
        match self.db.insert(key, value) {
            Ok(_) => {
                debug!(
                    "Cache put success: key_len={}, value_len={}",
                    key.len(),
                    value.len()
                );
                Ok(())
            }
            Err(e) => {
                error!("Cache put error: key_len={}, error={}", key.len(), e);
                Err(GuardianError::Store(format!("Cache put error: {}", e)))
            }
        }
    }

    #[instrument(level = "debug", skip(self, key))]
    async fn has(&self, key: &[u8]) -> Result<bool> {
        let _entered = self.span.enter();
        match self.db.contains_key(key) {
            Ok(exists) => {
                debug!("Cache has check: key_len={}, exists={}", key.len(), exists);
                Ok(exists)
            }
            Err(e) => {
                error!("Cache has error: key_len={}, error={}", key.len(), e);
                Err(GuardianError::Store(format!("Cache has error: {}", e)))
            }
        }
    }

    #[instrument(level = "debug", skip(self, key))]
    async fn delete(&self, key: &[u8]) -> Result<()> {
        let _entered = self.span.enter();
        match self.db.remove(key) {
            Ok(_) => {
                debug!("Cache delete success: key_len={}", key.len());
                Ok(())
            }
            Err(e) => {
                error!("Cache delete error: key_len={}, error={}", key.len(), e);
                Err(GuardianError::Store(format!("Cache delete error: {}", e)))
            }
        }
    }

    #[instrument(level = "debug", skip(self, query))]
    async fn query(&self, query: &crate::data_store::Query) -> Result<crate::data_store::Results> {
        let _entered = self.span.enter();
        use crate::data_store::{Key, ResultItem};

        debug!(
            "Cache query: has_prefix={}, limit={:?}, order={:?}",
            query.prefix.is_some(),
            query.limit,
            query.order
        );

        let iter: Box<dyn Iterator<Item = sled::Result<(sled::IVec, sled::IVec)>>> =
            if let Some(prefix_key) = &query.prefix {
                // Converte Key para bytes para usar como prefixo
                let prefix_bytes = prefix_key.as_bytes();
                Box::new(self.db.scan_prefix(prefix_bytes))
            } else {
                Box::new(self.db.iter())
            };

        let mut results = Vec::new();
        let mut count = 0;

        // Aplica offset se especificado
        let skip_count = query.offset.unwrap_or(0);
        let mut skipped = 0;

        for kv_result in iter {
            match kv_result {
                Ok((key_bytes, value_bytes)) => {
                    // Aplica offset
                    if skipped < skip_count {
                        skipped += 1;
                        continue;
                    }

                    // Converte bytes de volta para Key
                    let key_str = String::from_utf8_lossy(&key_bytes);
                    let key = Key::new(key_str.to_string());
                    let value = value_bytes.to_vec();

                    results.push(ResultItem::new(key, value));
                    count += 1;

                    // Aplica limite se especificado
                    if let Some(limit) = query.limit
                        && count >= limit
                    {
                        break;
                    }
                }
                Err(e) => {
                    error!("Cache query iteration error: error={}", e);
                    return Err(GuardianError::Store(format!("Cache query error: {}", e)));
                }
            }
        }

        // Aplica ordenação se necessário (Sled retorna em ordem ascendente por padrão)
        if matches!(query.order, crate::data_store::Order::Desc) {
            results.reverse();
        }

        debug!(
            "Cache query completed: results_count={}, skipped={}",
            results.len(),
            skipped
        );

        Ok(results)
    }

    #[instrument(level = "debug", skip(self, prefix))]
    async fn list_keys(&self, prefix: &[u8]) -> Result<Vec<crate::data_store::Key>> {
        let _entered = self.span.enter();
        use crate::data_store::Key;

        debug!("Cache list_keys: prefix_len={}", prefix.len());

        let mut keys = Vec::new();

        for kv_result in self.db.scan_prefix(prefix) {
            match kv_result {
                Ok((key_bytes, _)) => {
                    let key_str = String::from_utf8_lossy(&key_bytes);
                    let key = Key::new(key_str.to_string());
                    keys.push(key);
                }
                Err(e) => {
                    error!("Cache list_keys iteration error: error={}", e);
                    return Err(GuardianError::Store(format!(
                        "Cache list_keys error: {}",
                        e
                    )));
                }
            }
        }

        debug!("Cache list_keys completed: keys_count={}", keys.len());
        Ok(keys)
    }

    fn as_any(&self) -> &dyn std::any::Any {
        self
    }
}

/// Factory function para criar instâncias de cache
pub fn create_cache(opts: Options) -> SledCache {
    SledCache::new(opts)
}

/// Cria um cache padrão com configurações otimizadas
pub fn create_default_cache() -> SledCache {
    create_cache(Options::default())
}

/// Cria um cache em memória para testes
pub fn create_memory_cache() -> SledCache {
    create_cache(Options {
        cache_mode: CacheMode::InMemory,
        ..Default::default()
    })
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::data_store::{Key, Order, Query};

    #[tokio::test]
    async fn test_sled_datastore_basic_operations() {
        let datastore = SledDatastore::new(":memory:", Options::default()).unwrap();

        // Test put and get
        let key = b"test_key";
        let value = b"test_value";

        datastore.put(key, value).await.unwrap();
        let retrieved = datastore.get(key).await.unwrap();
        assert_eq!(retrieved, Some(value.to_vec()));

        // Test has
        assert!(datastore.has(key).await.unwrap());
        assert!(!datastore.has(b"non_existent").await.unwrap());

        // Test delete
        datastore.delete(key).await.unwrap();
        assert!(!datastore.has(key).await.unwrap());
        assert_eq!(datastore.get(key).await.unwrap(), None);
    }

    #[tokio::test]
    async fn test_sled_datastore_query() {
        let datastore = SledDatastore::new(":memory:", Options::default()).unwrap();

        // Insert test data
        datastore.put(b"/users/alice", b"alice_data").await.unwrap();
        datastore.put(b"/users/bob", b"bob_data").await.unwrap();
        datastore
            .put(b"/users/charlie", b"charlie_data")
            .await
            .unwrap();
        datastore
            .put(b"/config/database", b"db_config")
            .await
            .unwrap();

        // Test query with prefix
        let query = Query {
            prefix: Some(Key::new("/users")),
            limit: None,
            order: Order::Asc,
            offset: None,
        };

        let results = datastore.query(&query).await.unwrap();
        assert_eq!(results.len(), 3);

        // Test query with limit
        let query_limited = Query {
            prefix: Some(Key::new("/users")),
            limit: Some(2),
            order: Order::Asc,
            offset: None,
        };

        let results_limited = datastore.query(&query_limited).await.unwrap();
        assert_eq!(results_limited.len(), 2);

        // Test query with offset
        let query_offset = Query {
            prefix: Some(Key::new("/users")),
            limit: None,
            order: Order::Asc,
            offset: Some(1),
        };

        let results_offset = datastore.query(&query_offset).await.unwrap();
        assert_eq!(results_offset.len(), 2);
    }

    #[tokio::test]
    async fn test_sled_datastore_list_keys() {
        let datastore = SledDatastore::new(":memory:", Options::default()).unwrap();

        // Insert test data
        datastore.put(b"/users/alice", b"alice_data").await.unwrap();
        datastore.put(b"/users/bob", b"bob_data").await.unwrap();
        datastore
            .put(b"/config/database", b"db_config")
            .await
            .unwrap();

        // Test list_keys with prefix
        let keys = datastore.list_keys(b"/users").await.unwrap();
        assert_eq!(keys.len(), 2);

        let key_strings: Vec<String> = keys.iter().map(|k| k.as_str()).collect();
        assert!(key_strings.contains(&"/users/alice".to_string()));
        assert!(key_strings.contains(&"/users/bob".to_string()));
    }

    #[test]
    fn test_cache_mode_detection() {
        let persistent_opts = Options {
            cache_mode: CacheMode::Persistent,
            ..Default::default()
        };

        let memory_opts = Options {
            cache_mode: CacheMode::InMemory,
            ..Default::default()
        };

        assert_eq!(persistent_opts.cache_mode, CacheMode::Persistent);
        assert_eq!(memory_opts.cache_mode, CacheMode::InMemory);
    }
}