1
use std::borrow::Cow;
2
use std::collections::{HashMap, HashSet};
3
use std::fmt::{Debug, Display};
4
use std::fs::{self, File};
5
use std::io::{Read, Write};
6
use std::marker::PhantomData;
7
use std::path::{Path, PathBuf};
8
use std::sync::{Arc, Weak};
9

            
10
use bonsaidb_core::admin::database::{self, ByName, Database as DatabaseRecord};
11
use bonsaidb_core::admin::user::User;
12
use bonsaidb_core::admin::{self, Admin, PermissionGroup, Role, ADMIN_DATABASE_NAME};
13
use bonsaidb_core::circulate;
14
pub use bonsaidb_core::circulate::Relay;
15
use bonsaidb_core::connection::{
16
    self, Connection, HasSession, Identity, IdentityReference, LowLevelConnection, Session,
17
    SessionAuthentication, SessionId, StorageConnection,
18
};
19
use bonsaidb_core::document::CollectionDocument;
20
#[cfg(any(feature = "encryption", feature = "compression"))]
21
use bonsaidb_core::document::KeyId;
22
use bonsaidb_core::permissions::bonsai::{
23
    bonsaidb_resource_name, database_resource_name, role_resource_name, user_resource_name,
24
    BonsaiAction, ServerAction,
25
};
26
use bonsaidb_core::permissions::Permissions;
27
use bonsaidb_core::schema::{
28
    Nameable, NamedCollection, Schema, SchemaName, SchemaSummary, Schematic,
29
};
30
use fs2::FileExt;
31
use itertools::Itertools;
32
use nebari::io::any::{AnyFile, AnyFileManager};
33
use nebari::io::FileManager;
34
use nebari::{ChunkCache, ThreadPool};
35
use parking_lot::{Mutex, RwLock};
36
use rand::{thread_rng, Rng};
37

            
38
#[cfg(feature = "compression")]
39
use crate::config::Compression;
40
use crate::config::{KeyValuePersistence, StorageConfiguration};
41
use crate::database::Context;
42
use crate::tasks::manager::Manager;
43
use crate::tasks::TaskManager;
44
#[cfg(feature = "encryption")]
45
use crate::vault::{self, LocalVaultKeyStorage, Vault};
46
use crate::{Database, Error};
47

            
48
#[cfg(feature = "password-hashing")]
49
mod argon;
50
#[cfg(feature = "token-authentication")]
51
mod token_authentication;
52

            
53
mod backup;
54
mod pubsub;
55
pub use backup::{AnyBackupLocation, BackupLocation};
56

            
57
/// A file-based, multi-database, multi-user database engine. This type blocks
58
/// the current thread when used. See [`AsyncStorage`](crate::AsyncStorage) for
59
/// this type's async counterpart.
60
///
61
/// ## Converting between Blocking and Async Types
62
///
63
/// [`AsyncStorage`](crate::AsyncStorage) and [`Storage`] can be converted to
64
/// and from each other using:
65
///
66
/// - [`AsyncStorage::into_blocking()`](crate::AsyncStorage::into_blocking)
67
/// - [`AsyncStorage::to_blocking()`](crate::AsyncStorage::to_blocking)
68
/// - [`AsyncStorage::as_blocking()`](crate::AsyncStorage::as_blocking)
69
/// - [`Storage::into_async()`]
70
/// - [`Storage::to_async()`]
71
/// - [`Storage::into_async_with_runtime()`]
72
/// - [`Storage::to_async_with_runtime()`]
73
///
74
/// ## Converting from `Database::open` to `Storage::open`
75
///
76
/// [`Database::open`](Database::open) is a simple method that uses `Storage` to
77
/// create a database named `default` with the schema provided. These two ways
78
/// of opening the database are the same:
79
///
80
/// ```rust
81
/// // `bonsaidb_core` is re-exported to `bonsaidb::core` or `bonsaidb_local::core`.
82
/// use bonsaidb_core::connection::StorageConnection;
83
/// use bonsaidb_core::schema::Schema;
84
/// // `bonsaidb_local` is re-exported to `bonsaidb::local` if using the omnibus crate.
85
/// use bonsaidb_local::{
86
///     config::{Builder, StorageConfiguration},
87
///     Database, Storage,
88
/// };
89
/// # fn open<MySchema: Schema>() -> anyhow::Result<()> {
90
/// // This creates a Storage instance, creates a database, and returns it.
91
/// let db = Database::open::<MySchema>(StorageConfiguration::new("my-db.bonsaidb"))?;
92
///
93
/// // This is the equivalent code being executed:
94
/// let storage =
95
///     Storage::open(StorageConfiguration::new("my-db.bonsaidb").with_schema::<MySchema>()?)?;
96
/// storage.create_database::<MySchema>("default", true)?;
97
/// let db = storage.database::<MySchema>("default")?;
98
/// #     Ok(())
99
/// # }
100
/// ```
101
///
102
/// ## Using multiple databases
103
///
104
/// This example shows how to use `Storage` to create and use multiple databases
105
/// with multiple schemas:
106
///
107
/// ```rust
108
/// use bonsaidb_core::connection::StorageConnection;
109
/// use bonsaidb_core::schema::{Collection, Schema};
110
/// use bonsaidb_local::config::{Builder, StorageConfiguration};
111
/// use bonsaidb_local::Storage;
112
/// use serde::{Deserialize, Serialize};
113
///
114
/// #[derive(Debug, Schema)]
115
/// #[schema(name = "my-schema", collections = [BlogPost, Author])]
116
/// # #[schema(core = bonsaidb_core)]
117
/// struct MySchema;
118
///
119
/// #[derive(Debug, Serialize, Deserialize, Collection)]
120
/// #[collection(name = "blog-posts")]
121
/// # #[collection(core = bonsaidb_core)]
122
/// struct BlogPost {
123
///     pub title: String,
124
///     pub contents: String,
125
///     pub author_id: u64,
126
/// }
127
///
128
/// #[derive(Debug, Serialize, Deserialize, Collection)]
129
/// #[collection(name = "blog-posts")]
130
/// # #[collection(core = bonsaidb_core)]
131
/// struct Author {
132
///     pub name: String,
133
/// }
134
///
135
/// # fn open() -> anyhow::Result<()> {
136
/// let storage = Storage::open(
137
///     StorageConfiguration::new("my-db.bonsaidb")
138
///         .with_schema::<BlogPost>()?
139
///         .with_schema::<MySchema>()?,
140
/// )?;
141
///
142
/// storage.create_database::<BlogPost>("ectons-blog", true)?;
143
/// let ectons_blog = storage.database::<BlogPost>("ectons-blog")?;
144
/// storage.create_database::<MySchema>("another-db", true)?;
145
/// let another_db = storage.database::<MySchema>("another-db")?;
146
/// # Ok(())
147
/// # }
148
/// ```
149
11829669
#[derive(Debug, Clone)]
150
#[must_use]
151
pub struct Storage {
152
    pub(crate) instance: StorageInstance,
153
    pub(crate) authentication: Option<Arc<AuthenticatedSession>>,
154
    effective_session: Option<Arc<Session>>,
155
}
156

            
157
#[derive(Debug)]
158
pub struct AuthenticatedSession {
159
    // TODO: client_data,
160
    storage: Weak<Data>,
161
    pub session: Mutex<Session>,
162
}
163

            
164
5261
#[derive(Debug, Default)]
165
pub struct SessionSubscribers {
166
    pub subscribers: HashMap<u64, SessionSubscriber>,
167
    pub subscribers_by_session: HashMap<SessionId, HashSet<u64>>,
168
    pub last_id: u64,
169
}
170

            
171
impl SessionSubscribers {
172
    pub fn unregister(&mut self, subscriber_id: u64) {
173
846
        if let Some(session_id) = self
174
846
            .subscribers
175
846
            .remove(&subscriber_id)
176
846
            .and_then(|sub| sub.session_id)
177
        {
178
            if let Some(session_subscribers) = self.subscribers_by_session.get_mut(&session_id) {
179
                session_subscribers.remove(&subscriber_id);
180
            }
181
846
        }
182
846
    }
183
}
184

            
185
#[derive(Debug)]
186
pub struct SessionSubscriber {
187
    pub session_id: Option<SessionId>,
188
    pub subscriber: circulate::Subscriber,
189
}
190

            
191
impl Drop for AuthenticatedSession {
192
719
    fn drop(&mut self) {
193
719
        let mut session = self.session.lock();
194
719
        if let Some(id) = session.id.take() {
195
719
            if let Some(storage) = self.storage.upgrade() {
196
                // Deregister the session id once dropped.
197
304
                let mut sessions = storage.sessions.write();
198
304
                sessions.sessions.remove(&id);
199
304

            
200
304
                // Remove all subscribers.
201
304
                let mut sessions = storage.subscribers.write();
202
304
                for id in sessions
203
304
                    .subscribers_by_session
204
304
                    .remove(&id)
205
304
                    .into_iter()
206
304
                    .flatten()
207
                {
208
                    sessions.subscribers.remove(&id);
209
                }
210
415
            }
211
        }
212
719
    }
213
}
214

            
215
5261
#[derive(Debug, Default)]
216
struct AuthenticatedSessions {
217
    sessions: HashMap<SessionId, Arc<AuthenticatedSession>>,
218
    last_session_id: u64,
219
}
220

            
221
14601468
#[derive(Debug, Clone)]
222
pub struct StorageInstance {
223
    data: Arc<Data>,
224
}
225

            
226
impl From<StorageInstance> for Storage {
227
92169
    fn from(instance: StorageInstance) -> Self {
228
92169
        Self {
229
92169
            instance,
230
92169
            authentication: None,
231
92169
            effective_session: None,
232
92169
        }
233
92169
    }
234
}
235

            
236
struct Data {
237
    lock: StorageLock,
238
    path: PathBuf,
239
    parallelization: usize,
240
    threadpool: ThreadPool<AnyFile>,
241
    file_manager: AnyFileManager,
242
    pub(crate) tasks: TaskManager,
243
    schemas: RwLock<HashMap<SchemaName, Arc<dyn DatabaseOpener>>>,
244
    available_databases: RwLock<HashMap<String, SchemaName>>,
245
    open_roots: Mutex<HashMap<String, Context>>,
246
    // cfg check matches `Connection::authenticate`
247
    authenticated_permissions: Permissions,
248
    sessions: RwLock<AuthenticatedSessions>,
249
    pub(crate) subscribers: Arc<RwLock<SessionSubscribers>>,
250
    #[cfg(feature = "password-hashing")]
251
    argon: argon::Hasher,
252
    #[cfg(feature = "encryption")]
253
    pub(crate) vault: Arc<Vault>,
254
    #[cfg(feature = "encryption")]
255
    default_encryption_key: Option<KeyId>,
256
    #[cfg(any(feature = "compression", feature = "encryption"))]
257
    tree_vault: Option<TreeVault>,
258
    pub(crate) key_value_persistence: KeyValuePersistence,
259
    chunk_cache: ChunkCache,
260
    pub(crate) check_view_integrity_on_database_open: bool,
261
    relay: Relay,
262
}
263

            
264
impl Storage {
265
    /// Creates or opens a multi-database [`Storage`] with its data stored in `directory`.
266
5335
    pub fn open(configuration: StorageConfiguration) -> Result<Self, Error> {
267
5335
        let owned_path = configuration
268
5335
            .path
269
5335
            .clone()
270
5335
            .unwrap_or_else(|| PathBuf::from("db.bonsaidb"));
271
5335
        let file_manager = if configuration.memory_only {
272
71
            AnyFileManager::memory()
273
        } else {
274
5264
            AnyFileManager::std()
275
        };
276

            
277
5335
        let manager = Manager::default();
278
21340
        for _ in 0..configuration.workers.worker_count {
279
21340
            manager.spawn_worker();
280
21340
        }
281
5335
        let tasks = TaskManager::new(manager);
282
5335

            
283
5335
        fs::create_dir_all(&owned_path)?;
284

            
285
5335
        let storage_lock = Self::lookup_or_create_id(&configuration, &owned_path)?;
286

            
287
        #[cfg(feature = "encryption")]
288
5261
        let vault = {
289
5335
            let vault_key_storage = match configuration.vault_key_storage {
290
222
                Some(storage) => storage,
291
                None => Arc::new(
292
5113
                    LocalVaultKeyStorage::new(owned_path.join("vault-keys"))
293
5113
                        .map_err(|err| Error::Vault(vault::Error::Initializing(err.to_string())))?,
294
                ),
295
            };
296

            
297
5335
            Arc::new(Vault::initialize(
298
5335
                storage_lock.id(),
299
5335
                &owned_path,
300
5335
                vault_key_storage,
301
5335
            )?)
302
        };
303

            
304
5261
        let parallelization = configuration.workers.parallelization;
305
5261
        let check_view_integrity_on_database_open = configuration.views.check_integrity_on_open;
306
5261
        let key_value_persistence = configuration.key_value_persistence;
307
5261
        #[cfg(feature = "password-hashing")]
308
5261
        let argon = argon::Hasher::new(configuration.argon);
309
5261
        #[cfg(feature = "encryption")]
310
5261
        let default_encryption_key = configuration.default_encryption_key;
311
5261
        #[cfg(all(feature = "compression", feature = "encryption"))]
312
5261
        let tree_vault = TreeVault::new_if_needed(
313
5261
            default_encryption_key.clone(),
314
5261
            &vault,
315
5261
            configuration.default_compression,
316
5261
        );
317
5261
        #[cfg(all(not(feature = "compression"), feature = "encryption"))]
318
5261
        let tree_vault = TreeVault::new_if_needed(default_encryption_key.clone(), &vault);
319
5261
        #[cfg(all(feature = "compression", not(feature = "encryption")))]
320
5261
        let tree_vault = TreeVault::new_if_needed(configuration.default_compression);
321
5261

            
322
5261
        let authenticated_permissions = configuration.authenticated_permissions;
323
5261

            
324
5261
        let storage = Self {
325
5261
            instance: StorageInstance {
326
5261
                data: Arc::new(Data {
327
5261
                    lock: storage_lock,
328
5261
                    tasks,
329
5261
                    parallelization,
330
5261
                    subscribers: Arc::default(),
331
5261
                    authenticated_permissions,
332
5261
                    sessions: RwLock::default(),
333
5261
                    #[cfg(feature = "password-hashing")]
334
5261
                    argon,
335
5261
                    #[cfg(feature = "encryption")]
336
5261
                    vault,
337
5261
                    #[cfg(feature = "encryption")]
338
5261
                    default_encryption_key,
339
5261
                    #[cfg(any(feature = "compression", feature = "encryption"))]
340
5261
                    tree_vault,
341
5261
                    path: owned_path,
342
5261
                    file_manager,
343
5261
                    chunk_cache: ChunkCache::new(2000, 160_384),
344
5261
                    threadpool: ThreadPool::new(parallelization),
345
5261
                    schemas: RwLock::new(configuration.initial_schemas),
346
5261
                    available_databases: RwLock::default(),
347
5261
                    open_roots: Mutex::default(),
348
5261
                    key_value_persistence,
349
5261
                    check_view_integrity_on_database_open,
350
5261
                    relay: Relay::default(),
351
5261
                }),
352
5261
            },
353
5261
            authentication: None,
354
5261
            effective_session: None,
355
5261
        };
356
5261

            
357
5261
        storage.cache_available_databases()?;
358

            
359
5261
        storage.create_admin_database_if_needed()?;
360

            
361
5261
        Ok(storage)
362
5335
    }
363

            
364
    #[cfg(feature = "internal-apis")]
365
    #[doc(hidden)]
366
2631958
    pub fn database_without_schema(&self, name: &str) -> Result<Database, Error> {
367
2631958
        let name = name.to_owned();
368
2631958
        self.instance
369
2631958
            .database_without_schema(&name, Some(self), None)
370
2631958
    }
371

            
372
5335
    fn lookup_or_create_id(
373
5335
        configuration: &StorageConfiguration,
374
5335
        path: &Path,
375
5335
    ) -> Result<StorageLock, Error> {
376
5335
        let id_path = {
377
5335
            let storage_id = path.join("server-id");
378
5335
            if storage_id.exists() {
379
2
                storage_id
380
            } else {
381
5333
                path.join("storage-id")
382
            }
383
        };
384

            
385
5335
        let (id, file) = if let Some(id) = configuration.unique_id {
386
            // The configuraiton id override is not persisted to disk. This is
387
            // mostly to prevent someone from accidentally adding this
388
            // configuration, realizing it breaks things, and then wanting to
389
            // revert. This makes reverting to the old value easier.
390
            let file = if id_path.exists() {
391
                File::open(id_path)?
392
            } else {
393
                let mut file = File::create(id_path)?;
394
                let id = id.to_string();
395
                file.write_all(id.as_bytes())?;
396
                file
397
            };
398
            file.lock_exclusive()?;
399
            (id, file)
400
        } else {
401
            // Load/Store a randomly generated id into a file. While the value
402
            // is numerical, the file contents are the ascii decimal, making it
403
            // easier for a human to view, and if needed, edit.
404

            
405
5335
            if id_path.exists() {
406
                // This value is important enought to not allow launching the
407
                // server if the file can't be read or contains unexpected data.
408
710
                let mut file = File::open(id_path)?;
409
710
                file.lock_exclusive()?;
410
710
                let mut bytes = Vec::new();
411
710
                file.read_to_end(&mut bytes)?;
412
710
                let existing_id =
413
710
                    String::from_utf8(bytes).expect("server-id contains invalid data");
414
710

            
415
710
                (existing_id.parse().expect("server-id isn't numeric"), file)
416
            } else {
417
4625
                let id = { thread_rng().gen::<u64>() };
418
4625
                let mut file = File::create(id_path)?;
419
4625
                file.lock_exclusive()?;
420

            
421
4625
                file.write_all(id.to_string().as_bytes())?;
422

            
423
4625
                (id, file)
424
            }
425
        };
426
5335
        Ok(StorageLock::new(StorageId(id), file))
427
5335
    }
428

            
429
5261
    fn cache_available_databases(&self) -> Result<(), Error> {
430
5261
        let available_databases = self
431
5261
            .admin()
432
5261
            .view::<ByName>()
433
5261
            .query()?
434
5261
            .into_iter()
435
5261
            .map(|map| (map.key, map.value))
436
5261
            .collect();
437
5261
        let mut storage_databases = self.instance.data.available_databases.write();
438
5261
        *storage_databases = available_databases;
439
5261
        Ok(())
440
5261
    }
441

            
442
    fn create_admin_database_if_needed(&self) -> Result<(), Error> {
443
5261
        self.register_schema::<Admin>()?;
444
5261
        match self.database::<Admin>(ADMIN_DATABASE_NAME) {
445
636
            Ok(_) => {}
446
            Err(bonsaidb_core::Error::DatabaseNotFound(_)) => {
447
4625
                drop(self.create_database::<Admin>(ADMIN_DATABASE_NAME, true)?);
448
            }
449
            Err(err) => return Err(Error::Core(err)),
450
        }
451
5261
        Ok(())
452
5261
    }
453

            
454
    /// Returns the unique id of the server.
455
    ///
456
    /// This value is set from the [`StorageConfiguration`] or randomly
457
    /// generated when creating a server. It shouldn't be changed after a server
458
    /// is in use, as doing can cause issues. For example, the vault that
459
    /// manages encrypted storage uses the server ID to store the vault key. If
460
    /// the server ID changes, the vault key storage will need to be updated
461
    /// with the new server ID.
462
    #[must_use]
463
    pub fn unique_id(&self) -> StorageId {
464
        self.instance.data.lock.id()
465
    }
466

            
467
    #[must_use]
468
461075
    pub(crate) fn parallelization(&self) -> usize {
469
461075
        self.instance.data.parallelization
470
461075
    }
471

            
472
    #[must_use]
473
    #[cfg(feature = "encryption")]
474
2216293
    pub(crate) fn vault(&self) -> &Arc<Vault> {
475
2216293
        &self.instance.data.vault
476
2216293
    }
477

            
478
    #[must_use]
479
    #[cfg(any(feature = "encryption", feature = "compression"))]
480
4051890
    pub(crate) fn tree_vault(&self) -> Option<&TreeVault> {
481
4051890
        self.instance.data.tree_vault.as_ref()
482
4051890
    }
483

            
484
    #[must_use]
485
    #[cfg(feature = "encryption")]
486
3958632
    pub(crate) fn default_encryption_key(&self) -> Option<&KeyId> {
487
3958632
        self.instance.data.default_encryption_key.as_ref()
488
3958632
    }
489

            
490
    #[must_use]
491
    #[cfg(all(feature = "compression", not(feature = "encryption")))]
492
    #[allow(clippy::unused_self)]
493
    pub(crate) fn default_encryption_key(&self) -> Option<&KeyId> {
494
        None
495
    }
496

            
497
    /// Registers a schema for use within the server.
498
5261
    pub fn register_schema<DB: Schema>(&self) -> Result<(), Error> {
499
5261
        let mut schemas = self.instance.data.schemas.write();
500
5261
        if schemas
501
5261
            .insert(
502
5261
                DB::schema_name(),
503
5261
                Arc::new(StorageSchemaOpener::<DB>::new()?),
504
            )
505
5261
            .is_none()
506
        {
507
5261
            Ok(())
508
        } else {
509
            Err(Error::Core(bonsaidb_core::Error::SchemaAlreadyRegistered(
510
                DB::schema_name(),
511
            )))
512
        }
513
5261
    }
514

            
515
36293
    fn validate_name(name: &str) -> Result<(), Error> {
516
36293
        if name.chars().enumerate().all(|(index, c)| {
517
293344
            c.is_ascii_alphanumeric()
518
15321
                || (index == 0 && c == '_')
519
7216
                || (index > 0 && (c == '.' || c == '-'))
520
293344
        }) {
521
36139
            Ok(())
522
        } else {
523
154
            Err(Error::Core(bonsaidb_core::Error::InvalidDatabaseName(
524
154
                name.to_owned(),
525
154
            )))
526
        }
527
36293
    }
528

            
529
    /// Restricts an unauthenticated instance to having `effective_permissions`.
530
    /// Returns `None` if a session has already been established.
531
    #[must_use]
532
    pub fn with_effective_permissions(&self, effective_permissions: Permissions) -> Option<Self> {
533
        if self.effective_session.is_some() {
534
            None
535
        } else {
536
            Some(Self {
537
                instance: self.instance.clone(),
538
                authentication: self.authentication.clone(),
539
                effective_session: Some(Arc::new(Session {
540
                    id: None,
541
                    authentication: SessionAuthentication::None,
542
                    permissions: effective_permissions,
543
                })),
544
            })
545
        }
546
    }
547

            
548
    /// Converts this instance into its blocking version, which is able to be
549
    /// used without async. The returned instance uses the current Tokio runtime
550
    /// handle to spawn blocking tasks.
551
    ///
552
    /// # Panics
553
    ///
554
    /// Panics if called outside the context of a Tokio runtime.
555
    #[cfg(feature = "async")]
556
4481
    pub fn into_async(self) -> crate::AsyncStorage {
557
4481
        self.into_async_with_runtime(tokio::runtime::Handle::current())
558
4481
    }
559

            
560
    /// Converts this instance into its blocking version, which is able to be
561
    /// used without async. The returned instance uses the provided runtime
562
    /// handle to spawn blocking tasks.
563
    #[cfg(feature = "async")]
564
4481
    pub fn into_async_with_runtime(self, runtime: tokio::runtime::Handle) -> crate::AsyncStorage {
565
4481
        crate::AsyncStorage {
566
4481
            storage: self,
567
4481
            runtime: Arc::new(runtime),
568
4481
        }
569
4481
    }
570

            
571
    /// Converts this instance into its blocking version, which is able to be
572
    /// used without async. The returned instance uses the current Tokio runtime
573
    /// handle to spawn blocking tasks.
574
    ///
575
    /// # Panics
576
    ///
577
    /// Panics if called outside the context of a Tokio runtime.
578
    #[cfg(feature = "async")]
579
    pub fn to_async(&self) -> crate::AsyncStorage {
580
        self.clone().into_async()
581
    }
582

            
583
    /// Converts this instance into its blocking version, which is able to be
584
    /// used without async. The returned instance uses the provided runtime
585
    /// handle to spawn blocking tasks.
586
    #[cfg(feature = "async")]
587
    pub fn to_async_with_runtime(&self, runtime: tokio::runtime::Handle) -> crate::AsyncStorage {
588
        self.clone().into_async_with_runtime(runtime)
589
    }
590
}
591

            
592
impl Debug for Data {
593
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
594
        let mut f = f.debug_struct("Data");
595
        f.field("lock", &self.lock)
596
            .field("path", &self.path)
597
            .field("parallelization", &self.parallelization)
598
            .field("threadpool", &self.threadpool)
599
            .field("file_manager", &self.file_manager)
600
            .field("tasks", &self.tasks)
601
            .field("available_databases", &self.available_databases)
602
            .field("open_roots", &self.open_roots)
603
            .field("authenticated_permissions", &self.authenticated_permissions)
604
            .field("sessions", &self.sessions)
605
            .field("subscribers", &self.subscribers)
606
            .field("key_value_persistence", &self.key_value_persistence)
607
            .field("chunk_cache", &self.chunk_cache)
608
            .field(
609
                "check_view_integrity_on_database_open",
610
                &self.check_view_integrity_on_database_open,
611
            )
612
            .field("relay", &self.relay);
613

            
614
        if let Some(schemas) = self.schemas.try_read() {
615
            let mut schemas = schemas.keys().collect::<Vec<_>>();
616
            schemas.sort();
617
            f.field("schemas", &schemas);
618
        } else {
619
            f.field("schemas", &"RwLock locked");
620
        }
621

            
622
        #[cfg(feature = "password-hashing")]
623
        f.field("argon", &self.argon);
624
        #[cfg(feature = "encryption")]
625
        {
626
            f.field("vault", &self.vault)
627
                .field("default_encryption_key", &self.default_encryption_key);
628
        }
629
        #[cfg(any(feature = "compression", feature = "encryption"))]
630
        f.field("tree_vault", &self.tree_vault);
631

            
632
        f.finish()
633
    }
634
}
635

            
636
impl StorageInstance {
637
    #[cfg_attr(
638
        not(any(feature = "encryption", feature = "compression")),
639
        allow(unused_mut)
640
    )]
641
2724354
    pub(crate) fn open_roots(&self, name: &str) -> Result<Context, Error> {
642
2724354
        let mut open_roots = self.data.open_roots.lock();
643
2724354
        if let Some(roots) = open_roots.get(name) {
644
2688508
            Ok(roots.clone())
645
        } else {
646
35846
            let task_name = name.to_string();
647
35846

            
648
35846
            let mut config = nebari::Config::new(self.data.path.join(task_name))
649
35846
                .file_manager(self.data.file_manager.clone())
650
35846
                .cache(self.data.chunk_cache.clone())
651
35846
                .shared_thread_pool(&self.data.threadpool);
652

            
653
            #[cfg(any(feature = "encryption", feature = "compression"))]
654
35846
            if let Some(vault) = self.data.tree_vault.clone() {
655
9248
                config = config.vault(vault);
656
26860
            }
657

            
658
35846
            let roots = config.open().map_err(Error::from)?;
659
35846
            let context = Context::new(
660
35846
                roots,
661
35846
                self.data.key_value_persistence.clone(),
662
35846
                Some(self.data.lock.clone()),
663
35846
            );
664
35846

            
665
35846
            open_roots.insert(name.to_owned(), context.clone());
666
35846

            
667
35846
            Ok(context)
668
        }
669
2724354
    }
670

            
671
4479778
    pub(crate) fn tasks(&self) -> &'_ TaskManager {
672
4479778
        &self.data.tasks
673
4479778
    }
674

            
675
2724354
    pub(crate) fn check_view_integrity_on_database_open(&self) -> bool {
676
2724354
        self.data.check_view_integrity_on_database_open
677
2724354
    }
678

            
679
3924
    pub(crate) fn relay(&self) -> &'_ Relay {
680
3924
        &self.data.relay
681
3924
    }
682

            
683
    /// Opens a database through a generic-free trait.
684
2665276
    pub(crate) fn database_without_schema(
685
2665276
        &self,
686
2665276
        name: &str,
687
2665276
        storage: Option<&Storage>,
688
2665276
        expected_schema: Option<SchemaName>,
689
2665276
    ) -> Result<Database, Error> {
690
        // TODO switch to upgradable read now that we are on parking_lot
691
2660651
        let stored_schema = {
692
2665276
            let available_databases = self.data.available_databases.read();
693
2665276
            available_databases
694
2665276
                .get(name)
695
2665276
                .ok_or_else(|| {
696
4625
                    Error::Core(bonsaidb_core::Error::DatabaseNotFound(name.to_string()))
697
2665276
                })?
698
2660651
                .clone()
699
        };
700

            
701
2660651
        if let Some(expected_schema) = expected_schema {
702
28466
            if stored_schema != expected_schema {
703
                return Err(Error::Core(bonsaidb_core::Error::SchemaMismatch {
704
                    database_name: name.to_owned(),
705
                    schema: expected_schema,
706
                    stored_schema,
707
                }));
708
28466
            }
709
2632185
        }
710

            
711
2660651
        let mut schemas = self.data.schemas.write();
712
2660651
        let storage =
713
2660651
            storage.map_or_else(|| Cow::Owned(Storage::from(self.clone())), Cow::Borrowed);
714
2660651
        if let Some(schema) = schemas.get_mut(&stored_schema) {
715
2660651
            let db = schema.open(name.to_string(), storage.as_ref())?;
716
2660651
            Ok(db)
717
        } else {
718
            // The schema was stored, the user is requesting the same schema,
719
            // but it isn't registerd with the storage currently.
720
            Err(Error::Core(bonsaidb_core::Error::SchemaNotRegistered(
721
                stored_schema,
722
            )))
723
        }
724
2665276
    }
725

            
726
68
    fn update_user_with_named_id<
727
68
        'user,
728
68
        'other,
729
68
        Col: NamedCollection<PrimaryKey = u64>,
730
68
        U: Nameable<'user, u64> + Send + Sync,
731
68
        O: Nameable<'other, u64> + Send + Sync,
732
68
        F: FnOnce(&mut CollectionDocument<User>, u64) -> Result<bool, bonsaidb_core::Error>,
733
68
    >(
734
68
        &self,
735
68
        user: U,
736
68
        other: O,
737
68
        callback: F,
738
68
    ) -> Result<(), bonsaidb_core::Error> {
739
68
        let admin = self.admin();
740
68
        let other = other.name()?;
741
68
        let user = User::load(user.name()?, &admin)?;
742
68
        let other = other.id::<Col, _>(&admin)?;
743
68
        match (user, other) {
744
68
            (Some(mut user), Some(other)) => {
745
68
                if callback(&mut user, other)? {
746
34
                    user.update(&admin)?;
747
34
                }
748
68
                Ok(())
749
            }
750
            // TODO make this a generic not found with a name parameter.
751
            _ => Err(bonsaidb_core::Error::UserNotFound),
752
        }
753
68
    }
754

            
755
    #[cfg(any(feature = "token-authentication", feature = "password-hashing"))]
756
    #[cfg_attr(
757
        any(
758
            not(feature = "token-authentication"),
759
            not(feature = "password-hashing")
760
        ),
761
        allow(unused_variables, clippy::needless_pass_by_value)
762
    )]
763
867
    fn authenticate_inner(
764
867
        &self,
765
867
        authentication: bonsaidb_core::connection::Authentication,
766
867
        loaded_user: Option<CollectionDocument<User>>,
767
867
        current_session_id: Option<SessionId>,
768
867
        admin: &Database,
769
867
    ) -> Result<Storage, bonsaidb_core::Error> {
770
867
        use bonsaidb_core::connection::Authentication;
771
867
        match authentication {
772
            #[cfg(feature = "token-authentication")]
773
            Authentication::Token {
774
304
                id,
775
304
                now,
776
304
                now_hash,
777
304
                algorithm,
778
304
            } => self.begin_token_authentication(id, now, &now_hash, algorithm, admin),
779
            #[cfg(feature = "token-authentication")]
780
304
            Authentication::TokenChallengeResponse(hash) => {
781
304
                let session_id =
782
304
                    current_session_id.ok_or(bonsaidb_core::Error::InvalidCredentials)?;
783
304
                self.finish_token_authentication(session_id, &hash, admin)
784
            }
785
            #[cfg(feature = "password-hashing")]
786
259
            Authentication::Password { user, password } => {
787
259
                let user = match loaded_user {
788
259
                    Some(user) => user,
789
                    None => {
790
                        User::load(user, admin)?.ok_or(bonsaidb_core::Error::InvalidCredentials)?
791
                    }
792
                };
793
259
                let saved_hash = user
794
259
                    .contents
795
259
                    .argon_hash
796
259
                    .clone()
797
259
                    .ok_or(bonsaidb_core::Error::InvalidCredentials)?;
798

            
799
259
                self.data
800
259
                    .argon
801
259
                    .verify(user.header.id, password, saved_hash)?;
802
259
                self.assume_user(user, admin)
803
            }
804
        }
805
867
    }
806

            
807
411
    fn assume_user(
808
411
        &self,
809
411
        user: CollectionDocument<User>,
810
411
        admin: &Database,
811
411
    ) -> Result<Storage, bonsaidb_core::Error> {
812
411
        let permissions = user.contents.effective_permissions(
813
411
            admin,
814
411
            &admin.storage().instance.data.authenticated_permissions,
815
411
        )?;
816

            
817
411
        let mut sessions = self.data.sessions.write();
818
411
        sessions.last_session_id += 1;
819
411
        let session_id = SessionId(sessions.last_session_id);
820
411
        let session = Session {
821
411
            id: Some(session_id),
822
411
            authentication: SessionAuthentication::Identity(Arc::new(Identity::User {
823
411
                id: user.header.id,
824
411
                username: user.contents.username,
825
411
            })),
826
411
            permissions,
827
411
        };
828
411
        let authentication = Arc::new(AuthenticatedSession {
829
411
            storage: Arc::downgrade(&self.data),
830
411
            session: Mutex::new(session.clone()),
831
411
        });
832
411
        sessions.sessions.insert(session_id, authentication.clone());
833
411

            
834
411
        Ok(Storage {
835
411
            instance: self.clone(),
836
411
            authentication: Some(authentication),
837
411
            effective_session: Some(Arc::new(session)),
838
411
        })
839
411
    }
840

            
841
226
    fn assume_role(
842
226
        &self,
843
226
        role: CollectionDocument<Role>,
844
226
        admin: &Database,
845
226
    ) -> Result<Storage, bonsaidb_core::Error> {
846
226
        let permissions = role.contents.effective_permissions(
847
226
            admin,
848
226
            &admin.storage().instance.data.authenticated_permissions,
849
226
        )?;
850

            
851
226
        let mut sessions = self.data.sessions.write();
852
226
        sessions.last_session_id += 1;
853
226
        let session_id = SessionId(sessions.last_session_id);
854
226
        let session = Session {
855
226
            id: Some(session_id),
856
226
            authentication: SessionAuthentication::Identity(Arc::new(Identity::Role {
857
226
                id: role.header.id,
858
226
                name: role.contents.name,
859
226
            })),
860
226
            permissions,
861
226
        };
862
226
        let authentication = Arc::new(AuthenticatedSession {
863
226
            storage: Arc::downgrade(&self.data),
864
226
            session: Mutex::new(session.clone()),
865
226
        });
866
226
        sessions.sessions.insert(session_id, authentication.clone());
867
226

            
868
226
        Ok(Storage {
869
226
            instance: self.clone(),
870
226
            authentication: Some(authentication),
871
226
            effective_session: Some(Arc::new(session)),
872
226
        })
873
226
    }
874

            
875
452
    fn add_permission_group_to_user_inner(
876
452
        user: &mut CollectionDocument<User>,
877
452
        permission_group_id: u64,
878
452
    ) -> bool {
879
452
        if user.contents.groups.contains(&permission_group_id) {
880
226
            false
881
        } else {
882
226
            user.contents.groups.push(permission_group_id);
883
226
            true
884
        }
885
452
    }
886

            
887
304
    fn remove_permission_group_from_user_inner(
888
304
        user: &mut CollectionDocument<User>,
889
304
        permission_group_id: u64,
890
304
    ) -> bool {
891
304
        let old_len = user.contents.groups.len();
892
304
        user.contents.groups.retain(|id| id != &permission_group_id);
893
304
        old_len != user.contents.groups.len()
894
304
    }
895

            
896
304
    fn add_role_to_user_inner(user: &mut CollectionDocument<User>, role_id: u64) -> bool {
897
304
        if user.contents.roles.contains(&role_id) {
898
152
            false
899
        } else {
900
152
            user.contents.roles.push(role_id);
901
152
            true
902
        }
903
304
    }
904

            
905
304
    fn remove_role_from_user_inner(user: &mut CollectionDocument<User>, role_id: u64) -> bool {
906
304
        let old_len = user.contents.roles.len();
907
304
        user.contents.roles.retain(|id| id != &role_id);
908
304
        old_len != user.contents.roles.len()
909
304
    }
910
}
911

            
912
pub trait DatabaseOpener: Send + Sync {
913
    fn schematic(&self) -> &'_ Schematic;
914
    fn open(&self, name: String, storage: &Storage) -> Result<Database, Error>;
915
}
916

            
917
pub struct StorageSchemaOpener<DB: Schema> {
918
    schematic: Schematic,
919
    _phantom: PhantomData<DB>,
920
}
921

            
922
impl<DB> StorageSchemaOpener<DB>
923
where
924
    DB: Schema,
925
{
926
5650
    pub fn new() -> Result<Self, Error> {
927
5650
        let schematic = DB::schematic()?;
928
5650
        Ok(Self {
929
5650
            schematic,
930
5650
            _phantom: PhantomData,
931
5650
        })
932
5650
    }
933
}
934

            
935
impl<DB> DatabaseOpener for StorageSchemaOpener<DB>
936
where
937
    DB: Schema,
938
{
939
164
    fn schematic(&self) -> &'_ Schematic {
940
164
        &self.schematic
941
164
    }
942

            
943
78659
    fn open(&self, name: String, storage: &Storage) -> Result<Database, Error> {
944
78659
        let roots = storage.instance.open_roots(&name)?;
945
78659
        let db = Database::new::<DB, _>(name, roots, storage)?;
946
78659
        Ok(db)
947
78659
    }
948
}
949

            
950
impl HasSession for StorageInstance {
951
    fn session(&self) -> Option<&Session> {
952
        None
953
    }
954
}
955

            
956
impl StorageConnection for StorageInstance {
957
    type Authenticated = Storage;
958
    type Database = Database;
959

            
960
63703
    fn admin(&self) -> Self::Database {
961
63703
        Database::new::<Admin, _>(
962
63703
            ADMIN_DATABASE_NAME,
963
63703
            self.open_roots(ADMIN_DATABASE_NAME).unwrap(),
964
63703
            &Storage::from(self.clone()),
965
63703
        )
966
63703
        .unwrap()
967
63703
    }
968

            
969
    #[cfg_attr(feature = "tracing", tracing::instrument(
970
        level = "trace",
971
        skip(self, schema),
972
        fields(
973
            schema.authority = schema.authority.as_ref(),
974
            schema.name = schema.name.as_ref(),
975
        )
976
    ))]
977
    fn create_database_with_schema(
978
        &self,
979
        name: &str,
980
        schema: SchemaName,
981
        only_if_needed: bool,
982
    ) -> Result<(), bonsaidb_core::Error> {
983
36289
        Storage::validate_name(name)?;
984

            
985
        {
986
36137
            let schemas = self.data.schemas.read();
987
36137
            if !schemas.contains_key(&schema) {
988
152
                return Err(bonsaidb_core::Error::SchemaNotRegistered(schema));
989
35985
            }
990
35985
        }
991
35985

            
992
35985
        let mut available_databases = self.data.available_databases.write();
993
35985
        let admin = self.admin();
994
35985
        if !available_databases.contains_key(name) {
995
34861
            admin
996
34861
                .collection::<DatabaseRecord>()
997
34861
                .push(&admin::Database {
998
34861
                    name: name.to_string(),
999
34861
                    schema: schema.clone(),
34861
                })?;
34861
            available_databases.insert(name.to_string(), schema);
1124
        } else if !only_if_needed {
152
            return Err(bonsaidb_core::Error::DatabaseNameAlreadyTaken(
152
                name.to_string(),
152
            ));
972
        }

            
35833
        Ok(())
36289
    }

            
10771
    fn database<DB: Schema>(&self, name: &str) -> Result<Self::Database, bonsaidb_core::Error> {
10771
        self.database_without_schema(name, None, Some(DB::schema_name()))
10771
            .map_err(bonsaidb_core::Error::from)
10771
    }

            
    #[cfg_attr(feature = "tracing", tracing::instrument(level = "trace", skip(self)))]
18804
    fn delete_database(&self, name: &str) -> Result<(), bonsaidb_core::Error> {
18804
        let admin = self.admin();
18804
        let mut available_databases = self.data.available_databases.write();
18804
        available_databases.remove(name);
18804

            
18804
        let mut open_roots = self.data.open_roots.lock();
18804
        open_roots.remove(name);
18804

            
18804
        let database_folder = self.data.path.join(name);
18804
        if database_folder.exists() {
18543
            let file_manager = self.data.file_manager.clone();
18543
            file_manager
18543
                .delete_directory(&database_folder)
18543
                .map_err(Error::Nebari)?;
261
        }

            
18804
        if let Some(entry) = admin
18804
            .view::<database::ByName>()
18804
            .with_key(name)
18804
            .query()?
18804
            .first()
        {
18652
            admin.delete::<DatabaseRecord, _>(&entry.source)?;

            
18652
            Ok(())
        } else {
152
            Err(bonsaidb_core::Error::DatabaseNotFound(name.to_string()))
        }
18804
    }

            
    #[cfg_attr(feature = "tracing", tracing::instrument(level = "trace", skip_all))]
152
    fn list_databases(&self) -> Result<Vec<connection::Database>, bonsaidb_core::Error> {
152
        let available_databases = self.data.available_databases.read();
152
        Ok(available_databases
152
            .iter()
5595
            .map(|(name, schema)| connection::Database {
5595
                name: name.to_string(),
5595
                schema: schema.clone(),
5595
            })
152
            .collect())
152
    }

            
    #[cfg_attr(feature = "tracing", tracing::instrument(level = "trace", skip_all))]
152
    fn list_available_schemas(&self) -> Result<Vec<SchemaSummary>, bonsaidb_core::Error> {
152
        let available_databases = self.data.available_databases.read();
152
        let schemas = self.data.schemas.read();
152

            
152
        Ok(available_databases
152
            .values()
152
            .unique()
452
            .filter_map(|name| {
452
                schemas
452
                    .get(name)
452
                    .map(|opener| SchemaSummary::from(opener.schematic()))
452
            })
152
            .collect())
152
    }

            
    #[cfg_attr(feature = "tracing", tracing::instrument(level = "trace", skip_all))]
600
    fn create_user(&self, username: &str) -> Result<u64, bonsaidb_core::Error> {
600
        let result = self
600
            .admin()
600
            .collection::<User>()
600
            .push(&User::default_with_username(username))?;
563
        Ok(result.id)
600
    }

            
    #[cfg_attr(feature = "tracing", tracing::instrument(level = "trace", skip_all))]
8
    fn delete_user<'user, U: Nameable<'user, u64> + Send + Sync>(
8
        &self,
8
        user: U,
8
    ) -> Result<(), bonsaidb_core::Error> {
8
        let admin = self.admin();
8
        let user = User::load(user, &admin)?.ok_or(bonsaidb_core::Error::UserNotFound)?;
8
        user.delete(&admin)?;

            
8
        Ok(())
8
    }

            
    #[cfg(feature = "password-hashing")]
    #[cfg_attr(feature = "tracing", tracing::instrument(level = "trace", skip_all))]
185
    fn set_user_password<'user, U: Nameable<'user, u64> + Send + Sync>(
185
        &self,
185
        user: U,
185
        password: bonsaidb_core::connection::SensitiveString,
185
    ) -> Result<(), bonsaidb_core::Error> {
185
        let admin = self.admin();
185
        let mut user = User::load(user, &admin)?.ok_or(bonsaidb_core::Error::UserNotFound)?;
185
        user.contents.argon_hash = Some(self.data.argon.hash(user.header.id, password)?);
185
        user.update(&admin)
185
    }

            
    #[cfg_attr(feature = "tracing", tracing::instrument(level = "trace", skip_all))]
    #[cfg(any(feature = "token-authentication", feature = "password-hashing"))]
    fn authenticate(
        &self,
        authentication: bonsaidb_core::connection::Authentication,
    ) -> Result<Self::Authenticated, bonsaidb_core::Error> {
        let admin = self.admin();
        self.authenticate_inner(authentication, None, None, &admin)
            .map(Storage::from)
    }

            
    #[cfg_attr(feature = "tracing", tracing::instrument(level = "trace", skip_all))]
    fn assume_identity(
        &self,
        identity: IdentityReference<'_>,
    ) -> Result<Self::Authenticated, bonsaidb_core::Error> {
        let admin = self.admin();
        match identity {
            IdentityReference::User(user) => {
                let user =
                    User::load(user, &admin)?.ok_or(bonsaidb_core::Error::InvalidCredentials)?;
                self.assume_user(user, &admin).map(Storage::from)
            }
            IdentityReference::Role(role) => {
                let role =
                    Role::load(role, &admin)?.ok_or(bonsaidb_core::Error::InvalidCredentials)?;
                self.assume_role(role, &admin).map(Storage::from)
            }
            _ => Err(bonsaidb_core::Error::InvalidCredentials),
        }
    }

            
    #[cfg_attr(feature = "tracing", tracing::instrument(level = "trace", skip_all))]
    fn add_permission_group_to_user<
        'user,
        'group,
        U: Nameable<'user, u64> + Send + Sync,
        G: Nameable<'group, u64> + Send + Sync,
    >(
        &self,
        user: U,
        permission_group: G,
    ) -> Result<(), bonsaidb_core::Error> {
        self.update_user_with_named_id::<PermissionGroup, _, _, _>(
            user,
            permission_group,
            |user, permission_group_id| {
                Ok(Self::add_permission_group_to_user_inner(
                    user,
                    permission_group_id,
                ))
            },
        )
    }

            
    #[cfg_attr(feature = "tracing", tracing::instrument(level = "trace", skip_all))]
    fn remove_permission_group_from_user<
        'user,
        'group,
        U: Nameable<'user, u64> + Send + Sync,
        G: Nameable<'group, u64> + Send + Sync,
    >(
        &self,
        user: U,
        permission_group: G,
    ) -> Result<(), bonsaidb_core::Error> {
        self.update_user_with_named_id::<PermissionGroup, _, _, _>(
            user,
            permission_group,
            |user, permission_group_id| {
                Ok(Self::remove_permission_group_from_user_inner(
                    user,
                    permission_group_id,
                ))
            },
        )
    }

            
    #[cfg_attr(feature = "tracing", tracing::instrument(level = "trace", skip_all))]
    fn add_role_to_user<
        'user,
        'group,
        U: Nameable<'user, u64> + Send + Sync,
        G: Nameable<'group, u64> + Send + Sync,
    >(
        &self,
        user: U,
        role: G,
    ) -> Result<(), bonsaidb_core::Error> {
        self.update_user_with_named_id::<PermissionGroup, _, _, _>(user, role, |user, role_id| {
            Ok(Self::add_role_to_user_inner(user, role_id))
        })
    }

            
    #[cfg_attr(feature = "tracing", tracing::instrument(level = "trace", skip_all))]
    fn remove_role_from_user<
        'user,
        'group,
        U: Nameable<'user, u64> + Send + Sync,
        G: Nameable<'group, u64> + Send + Sync,
    >(
        &self,
        user: U,
        role: G,
    ) -> Result<(), bonsaidb_core::Error> {
        self.update_user_with_named_id::<Role, _, _, _>(user, role, |user, role_id| {
            Ok(Self::remove_role_from_user_inner(user, role_id))
        })
    }
}

            
impl HasSession for Storage {
5389404
    fn session(&self) -> Option<&Session> {
5389404
        self.effective_session.as_deref()
5389404
    }
}

            
impl StorageConnection for Storage {
    type Authenticated = Self;
    type Database = Database;

            
6613
    fn admin(&self) -> Self::Database {
6613
        self.instance.admin()
6613
    }

            
    fn create_database_with_schema(
        &self,
        name: &str,
        schema: SchemaName,
        only_if_needed: bool,
    ) -> Result<(), bonsaidb_core::Error> {
36289
        self.check_permission(
36289
            database_resource_name(name),
36289
            &BonsaiAction::Server(ServerAction::CreateDatabase),
36289
        )?;
36289
        self.instance
36289
            .create_database_with_schema(name, schema, only_if_needed)
36289
    }

            
10771
    fn database<DB: Schema>(&self, name: &str) -> Result<Self::Database, bonsaidb_core::Error> {
10771
        self.instance.database::<DB>(name)
10771
    }

            
    fn delete_database(&self, name: &str) -> Result<(), bonsaidb_core::Error> {
18804
        self.check_permission(
18804
            database_resource_name(name),
18804
            &BonsaiAction::Server(ServerAction::DeleteDatabase),
18804
        )?;
18804
        self.instance.delete_database(name)
18804
    }

            
    fn list_databases(&self) -> Result<Vec<connection::Database>, bonsaidb_core::Error> {
152
        self.check_permission(
152
            bonsaidb_resource_name(),
152
            &BonsaiAction::Server(ServerAction::ListDatabases),
152
        )?;
152
        self.instance.list_databases()
152
    }

            
    fn list_available_schemas(&self) -> Result<Vec<SchemaSummary>, bonsaidb_core::Error> {
152
        self.check_permission(
152
            bonsaidb_resource_name(),
152
            &BonsaiAction::Server(ServerAction::ListAvailableSchemas),
152
        )?;
152
        self.instance.list_available_schemas()
152
    }

            
    fn create_user(&self, username: &str) -> Result<u64, bonsaidb_core::Error> {
637
        self.check_permission(
637
            bonsaidb_resource_name(),
637
            &BonsaiAction::Server(ServerAction::CreateUser),
637
        )?;
600
        self.instance.create_user(username)
637
    }

            
8
    fn delete_user<'user, U: Nameable<'user, u64> + Send + Sync>(
8
        &self,
8
        user: U,
8
    ) -> Result<(), bonsaidb_core::Error> {
8
        let admin = self.admin();
8
        let user = user.name()?;
8
        let user_id = user
8
            .id::<User, _>(&admin)?
8
            .ok_or(bonsaidb_core::Error::UserNotFound)?;
8
        self.check_permission(
8
            user_resource_name(user_id),
8
            &BonsaiAction::Server(ServerAction::DeleteUser),
8
        )?;
8
        self.instance.delete_user(user)
8
    }

            
    #[cfg(feature = "password-hashing")]
5
    fn set_user_password<'user, U: Nameable<'user, u64> + Send + Sync>(
5
        &self,
5
        user: U,
5
        password: bonsaidb_core::connection::SensitiveString,
5
    ) -> Result<(), bonsaidb_core::Error> {
5
        let admin = self.admin();
5
        let user = user.name()?;
5
        let user_id = user
5
            .id::<User, _>(&admin)?
5
            .ok_or(bonsaidb_core::Error::UserNotFound)?;
5
        self.check_permission(
5
            user_resource_name(user_id),
5
            &BonsaiAction::Server(ServerAction::SetPassword),
5
        )?;
5
        self.instance.set_user_password(user, password)
5
    }

            
    #[cfg(any(feature = "token-authentication", feature = "password-hashing"))]
    #[cfg_attr(not(feature = "token-authentication"), allow(unused_assignments))]
    #[cfg_attr(not(feature = "password-hashing"), allow(unused_mut))]
867
    fn authenticate(
867
        &self,
867
        authentication: bonsaidb_core::connection::Authentication,
867
    ) -> Result<Self, bonsaidb_core::Error> {
867
        let admin = self.admin();
867
        let mut loaded_user = None;
867
        match &authentication {
            #[cfg(feature = "token-authentication")]
304
            bonsaidb_core::connection::Authentication::Token { id, .. } => {
304
                self.check_permission(
304
                    bonsaidb_core::permissions::bonsai::authentication_token_resource_name(*id),
304
                    &BonsaiAction::Server(ServerAction::Authenticate(
304
                        bonsaidb_core::connection::AuthenticationMethod::Token,
304
                    )),
304
                )?;
            }
            #[cfg(feature = "password-hashing")]
259
            bonsaidb_core::connection::Authentication::Password { user, .. } => {
259
                let user =
259
                    User::load(user, &admin)?.ok_or(bonsaidb_core::Error::InvalidCredentials)?;
259
                self.check_permission(
259
                    user_resource_name(user.header.id),
259
                    &BonsaiAction::Server(ServerAction::Authenticate(
259
                        bonsaidb_core::connection::AuthenticationMethod::PasswordHash,
259
                    )),
259
                )?;
259
                loaded_user = Some(user);
            }
            #[cfg(feature = "token-authentication")]
304
            bonsaidb_core::connection::Authentication::TokenChallengeResponse(_) => {}
        }
867
        self.instance.authenticate_inner(
867
            authentication,
867
            loaded_user,
867
            self.authentication
867
                .as_ref()
867
                .and_then(|auth| auth.session.lock().id),
867
            &admin,
867
        )
867
    }

            
74
    fn assume_identity(
74
        &self,
74
        identity: IdentityReference<'_>,
74
    ) -> Result<Self::Authenticated, bonsaidb_core::Error> {
74
        match identity {
            IdentityReference::User(user) => {
                let admin = self.admin();
                let user =
                    User::load(user, &admin)?.ok_or(bonsaidb_core::Error::InvalidCredentials)?;
                self.check_permission(
                    user_resource_name(user.header.id),
                    &BonsaiAction::Server(ServerAction::AssumeIdentity),
                )?;
                self.instance.assume_user(user, &admin)
            }
74
            IdentityReference::Role(role) => {
74
                let admin = self.admin();
74
                let role =
74
                    Role::load(role, &admin)?.ok_or(bonsaidb_core::Error::InvalidCredentials)?;
74
                self.check_permission(
74
                    role_resource_name(role.header.id),
74
                    &BonsaiAction::Server(ServerAction::AssumeIdentity),
74
                )?;
74
                self.instance.assume_role(role, &admin)
            }

            
            _ => Err(bonsaidb_core::Error::InvalidCredentials),
        }
74
    }

            
20
    fn add_permission_group_to_user<
20
        'user,
20
        'group,
20
        U: Nameable<'user, u64> + Send + Sync,
20
        G: Nameable<'group, u64> + Send + Sync,
20
    >(
20
        &self,
20
        user: U,
20
        permission_group: G,
20
    ) -> Result<(), bonsaidb_core::Error> {
20
        self.instance
20
            .update_user_with_named_id::<PermissionGroup, _, _, _>(
20
                user,
20
                permission_group,
20
                |user, permission_group_id| {
20
                    self.check_permission(
20
                        user_resource_name(user.header.id),
20
                        &BonsaiAction::Server(ServerAction::ModifyUserPermissionGroups),
20
                    )?;
20
                    Ok(StorageInstance::add_permission_group_to_user_inner(
20
                        user,
20
                        permission_group_id,
20
                    ))
20
                },
20
            )
20
    }

            
16
    fn remove_permission_group_from_user<
16
        'user,
16
        'group,
16
        U: Nameable<'user, u64> + Send + Sync,
16
        G: Nameable<'group, u64> + Send + Sync,
16
    >(
16
        &self,
16
        user: U,
16
        permission_group: G,
16
    ) -> Result<(), bonsaidb_core::Error> {
16
        self.instance
16
            .update_user_with_named_id::<PermissionGroup, _, _, _>(
16
                user,
16
                permission_group,
16
                |user, permission_group_id| {
16
                    self.check_permission(
16
                        user_resource_name(user.header.id),
16
                        &BonsaiAction::Server(ServerAction::ModifyUserPermissionGroups),
16
                    )?;
16
                    Ok(StorageInstance::remove_permission_group_from_user_inner(
16
                        user,
16
                        permission_group_id,
16
                    ))
16
                },
16
            )
16
    }

            
16
    fn add_role_to_user<
16
        'user,
16
        'group,
16
        U: Nameable<'user, u64> + Send + Sync,
16
        G: Nameable<'group, u64> + Send + Sync,
16
    >(
16
        &self,
16
        user: U,
16
        role: G,
16
    ) -> Result<(), bonsaidb_core::Error> {
16
        self.instance
16
            .update_user_with_named_id::<PermissionGroup, _, _, _>(user, role, |user, role_id| {
16
                self.check_permission(
16
                    user_resource_name(user.header.id),
16
                    &BonsaiAction::Server(ServerAction::ModifyUserRoles),
16
                )?;
16
                Ok(StorageInstance::add_role_to_user_inner(user, role_id))
16
            })
16
    }

            
16
    fn remove_role_from_user<
16
        'user,
16
        'group,
16
        U: Nameable<'user, u64> + Send + Sync,
16
        G: Nameable<'group, u64> + Send + Sync,
16
    >(
16
        &self,
16
        user: U,
16
        role: G,
16
    ) -> Result<(), bonsaidb_core::Error> {
16
        self.instance
16
            .update_user_with_named_id::<Role, _, _, _>(user, role, |user, role_id| {
16
                self.check_permission(
16
                    user_resource_name(user.header.id),
16
                    &BonsaiAction::Server(ServerAction::ModifyUserRoles),
16
                )?;
16
                Ok(StorageInstance::remove_role_from_user_inner(user, role_id))
16
            })
16
    }
}

            
1
#[test]
1
fn name_validation_tests() {
1
    assert!(matches!(Storage::validate_name("azAZ09.-"), Ok(())));
1
    assert!(matches!(
1
        Storage::validate_name("_internal-names-work"),
        Ok(())
    ));
1
    assert!(matches!(
1
        Storage::validate_name("-alphaunmericfirstrequired"),
        Err(Error::Core(bonsaidb_core::Error::InvalidDatabaseName(_)))
    ));
1
    assert!(matches!(
1
        Storage::validate_name("\u{2661}"),
        Err(Error::Core(bonsaidb_core::Error::InvalidDatabaseName(_)))
    ));
1
}

            
/// The unique id of a [`Storage`] instance.
35846
#[derive(Clone, Copy, Eq, PartialEq, Hash)]
pub struct StorageId(u64);

            
impl StorageId {
    /// Returns the id as a u64.
    #[must_use]
    pub const fn as_u64(self) -> u64 {
        self.0
    }
}

            
impl Debug for StorageId {
9961
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
9961
        // let formatted_length = format!();
9961
        write!(f, "{:016x}", self.0)
9961
    }
}

            
impl Display for StorageId {
9961
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
9961
        Debug::fmt(self, f)
9961
    }
}

            
2672112
#[derive(Debug, Clone)]
#[cfg(any(feature = "compression", feature = "encryption"))]
pub(crate) struct TreeVault {
    #[cfg(feature = "compression")]
    compression: Option<Compression>,
    #[cfg(feature = "encryption")]
    pub key: Option<KeyId>,
    #[cfg(feature = "encryption")]
    pub vault: Arc<Vault>,
}

            
#[cfg(all(feature = "compression", feature = "encryption"))]
impl TreeVault {
2221554
    pub(crate) fn new_if_needed(
2221554
        key: Option<KeyId>,
2221554
        vault: &Arc<Vault>,
2221554
        compression: Option<Compression>,
2221554
    ) -> Option<Self> {
2221554
        if key.is_none() && compression.is_none() {
2170752
            None
        } else {
50802
            Some(Self {
50802
                key,
50802
                compression,
50802
                vault: vault.clone(),
50802
            })
        }
2221554
    }

            
8118744
    fn header(&self, compressed: bool) -> u8 {
8118744
        let mut bits = if self.key.is_some() { 0b1000_0000 } else { 0 };

            
8118744
        if compressed {
3869784
            if let Some(compression) = self.compression {
3869784
                bits |= compression as u8;
3869784
            }
4248960
        }

            
8118744
        bits
8118744
    }
}

            
#[cfg(all(feature = "compression", feature = "encryption"))]
impl nebari::Vault for TreeVault {
    type Error = Error;

            
8118744
    fn encrypt(&self, payload: &[u8]) -> Result<Vec<u8>, Error> {
8118744
        // TODO this allocates too much. The vault should be able to do an
8118744
        // in-place encryption operation so that we can use a single buffer.
8118744
        let mut includes_compression = false;
8118744
        let compressed = match (payload.len(), self.compression) {
3874373
            (128..=usize::MAX, Some(Compression::Lz4)) => {
3869784
                includes_compression = true;
3869784
                Cow::Owned(lz4_flex::block::compress_prepend_size(payload))
            }
4248960
            _ => Cow::Borrowed(payload),
        };

            
8118744
        let mut complete = if let Some(key) = &self.key {
74318
            self.vault.encrypt_payload(key, &compressed, None)?
        } else {
8044426
            compressed.into_owned()
        };

            
8118744
        let header = self.header(includes_compression);
8118744
        if header != 0 {
3937688
            let header = [b't', b'r', b'v', header];
3937688
            complete.splice(0..0, header);
4181056
        }

            
8118744
        Ok(complete)
8118744
    }

            
904401
    fn decrypt(&self, payload: &[u8]) -> Result<Vec<u8>, Error> {
904401
        if payload.len() >= 4 && &payload[0..3] == b"trv" {
755685
            let header = payload[3];
755685
            let payload = &payload[4..];
755685
            let encrypted = (header & 0b1000_0000) != 0;
755685
            let compression = header & 0b0111_1111;
755685
            let decrypted = if encrypted {
15980
                Cow::Owned(self.vault.decrypt_payload(payload, None)?)
            } else {
739705
                Cow::Borrowed(payload)
            };
            #[allow(clippy::single_match)] // Make it an error when we add a new algorithm
755684
            return Ok(match Compression::from_u8(compression) {
                Some(Compression::Lz4) => {
744368
                    lz4_flex::block::decompress_size_prepended(&decrypted).map_err(Error::from)?
                }
11316
                None => decrypted.into_owned(),
            });
148716
        }
148716
        self.vault.decrypt_payload(payload, None)
904401
    }
}

            
/// Functionality that is available on both [`Storage`] and
/// [`AsyncStorage`](crate::AsyncStorage).
pub trait StorageNonBlocking: Sized {
    /// Returns the path of the database storage.
    #[must_use]
    fn path(&self) -> &Path;

            
    /// Returns a new instance of [`Storage`] with `session` as the effective
    /// authentication session. This call will only succeed if there is no
    /// current session.
    fn assume_session(&self, session: Session) -> Result<Self, bonsaidb_core::Error>;
}

            
impl StorageNonBlocking for Storage {
6216
    fn path(&self) -> &Path {
6216
        &self.instance.data.path
6216
    }

            
2678837
    fn assume_session(&self, session: Session) -> Result<Storage, bonsaidb_core::Error> {
2678837
        if self.authentication.is_some() {
            // TODO better error
            return Err(bonsaidb_core::Error::InvalidCredentials);
2678837
        }

            
2678837
        let Some(session_id) = session.id else {
2677875
            return Ok(Self {
2677875
                instance: self.instance.clone(),
2677875
                authentication: None,
2677875
                effective_session: Some(Arc::new(session)),
2677875
            });
        };

            
962
        let session_data = self.instance.data.sessions.read();
        // TODO better error
962
        let authentication = session_data
962
            .sessions
962
            .get(&session_id)
962
            .ok_or(bonsaidb_core::Error::InvalidCredentials)?;

            
814
        let authentication_session = authentication.session.lock();
814
        let effective_permissions =
814
            Permissions::merged([&session.permissions, &authentication_session.permissions]);
814
        let effective_session = Session {
814
            id: authentication_session.id,
814
            authentication: authentication_session.authentication.clone(),
814
            permissions: effective_permissions,
814
        };
814

            
814
        Ok(Self {
814
            instance: self.instance.clone(),
814
            authentication: Some(authentication.clone()),
814
            effective_session: Some(Arc::new(effective_session)),
814
        })
2678837
    }
}

            
#[cfg(all(feature = "compression", not(feature = "encryption")))]
impl TreeVault {
    pub(crate) fn new_if_needed(compression: Option<Compression>) -> Option<Self> {
        compression.map(|compression| Self {
            compression: Some(compression),
        })
    }
}

            
#[cfg(all(feature = "compression", not(feature = "encryption")))]
impl nebari::Vault for TreeVault {
    type Error = Error;

            
    fn encrypt(&self, payload: &[u8]) -> Result<Vec<u8>, Error> {
        Ok(match (payload.len(), self.compression) {
            (128..=usize::MAX, Some(Compression::Lz4)) => {
                let mut destination =
                    vec![0; lz4_flex::block::get_maximum_output_size(payload.len()) + 8];
                let compressed_length =
                    lz4_flex::block::compress_into(payload, &mut destination[8..])
                        .expect("lz4-flex documents this shouldn't fail");
                destination.truncate(compressed_length + 8);
                destination[0..4].copy_from_slice(&[b't', b'r', b'v', Compression::Lz4 as u8]);
                // to_le_bytes() makes it compatible with lz4-flex decompress_size_prepended.
                let uncompressed_length =
                    u32::try_from(payload.len()).expect("nebari doesn't support >32 bit blocks");
                destination[4..8].copy_from_slice(&uncompressed_length.to_le_bytes());
                destination
            }
            // TODO this shouldn't copy
            _ => payload.to_vec(),
        })
    }

            
    fn decrypt(&self, payload: &[u8]) -> Result<Vec<u8>, Error> {
        if payload.len() >= 4 && &payload[0..3] == b"trv" {
            let header = payload[3];
            let payload = &payload[4..];
            let encrypted = (header & 0b1000_0000) != 0;
            let compression = header & 0b0111_1111;
            if encrypted {
                return Err(Error::EncryptionDisabled);
            }

            
            #[allow(clippy::single_match)] // Make it an error when we add a new algorithm
            return Ok(match Compression::from_u8(compression) {
                Some(Compression::Lz4) => {
                    lz4_flex::block::decompress_size_prepended(payload).map_err(Error::from)?
                }
                None => payload.to_vec(),
            });
        }
        Ok(payload.to_vec())
    }
}

            
#[cfg(all(not(feature = "compression"), feature = "encryption"))]
impl TreeVault {
    pub(crate) fn new_if_needed(key: Option<KeyId>, vault: &Arc<Vault>) -> Option<Self> {
        key.map(|key| Self {
            key: Some(key),
            vault: vault.clone(),
        })
    }

            
    #[allow(dead_code)] // This implementation is sort of documentation for what it would be. But our Vault payload already can detect if a parsing error occurs, so we don't need a header if only encryption is enabled.
    fn header(&self) -> u8 {
        if self.key.is_some() {
            0b1000_0000
        } else {
            0
        }
    }
}

            
#[cfg(all(not(feature = "compression"), feature = "encryption"))]
impl nebari::Vault for TreeVault {
    type Error = Error;

            
    fn encrypt(&self, payload: &[u8]) -> Result<Vec<u8>, Error> {
        if let Some(key) = &self.key {
            self.vault.encrypt_payload(key, payload, None)
        } else {
            // TODO does this need to copy?
            Ok(payload.to_vec())
        }
    }

            
    fn decrypt(&self, payload: &[u8]) -> Result<Vec<u8>, Error> {
        self.vault.decrypt_payload(payload, None)
    }
}

            
35846
#[derive(Clone, Debug)]
pub struct StorageLock(StorageId, Arc<LockData>);

            
impl StorageLock {
5335
    pub const fn id(&self) -> StorageId {
5335
        self.0
5335
    }
}

            
#[derive(Debug)]
struct LockData(File);

            
impl StorageLock {
5335
    fn new(id: StorageId, file: File) -> Self {
5335
        Self(id, Arc::new(LockData(file)))
5335
    }
}

            
impl Drop for LockData {
5187
    fn drop(&mut self) {
5187
        drop(self.0.unlock());
5187
    }
}