#[derive(Debug, Clone)]

    pub async fn open(configuration: StorageConfiguration) -> Result<Self, Error> {

        let owned_path = configuration

            .path

            .clone()

            .unwrap_or_else(|| PathBuf::from("db.bonsaidb"));

        let file_manager = if configuration.memory_only {

            AnyFileManager::memory()

            AnyFileManager::std()

        let manager = Manager::default();

        for _ in 0..configuration.workers.worker_count {

            manager.spawn_worker();

        }

        let tasks = TaskManager::new(manager);

        fs::create_dir_all(&owned_path).await?;

        let id = Self::lookup_or_create_id(&configuration, &owned_path).await?;

        let vault = {

            let vault_key_storage = match configuration.vault_key_storage {

                Some(storage) => storage,

                    LocalVaultKeyStorage::new(owned_path.join("vault-keys"))

                        .await

                        .map_err(|err| Error::Vault(vault::Error::Initializing(err.to_string())))?,

            Arc::new(Vault::initialize(id, &owned_path, vault_key_storage).await?)

        let parallelization = configuration.workers.parallelization;

        let check_view_integrity_on_database_open = configuration.views.check_integrity_on_open;

        let key_value_persistence = configuration.key_value_persistence;

        #[cfg(feature = "password-hashing")]

        let argon = argon::Hasher::new(configuration.argon);

        #[cfg(feature = "encryption")]

        let default_encryption_key = configuration.default_encryption_key;

        #[cfg(all(feature = "compression", feature = "encryption"))]

        let tree_vault = TreeVault::new_if_needed(

            default_encryption_key.clone(),

            &vault,

            configuration.default_compression,

        );

        let storage = tokio::task::spawn_blocking::<_, Result<Self, Error>>(move || {

            Ok(Self {

                data: Arc::new(Data {

                    id,

                    tasks,

                    parallelization,

                    #[cfg(feature = "password-hashing")]

                    argon,

                    #[cfg(feature = "encryption")]

                    vault,

                    #[cfg(feature = "encryption")]

                    default_encryption_key,

                    #[cfg(any(feature = "compression", feature = "encryption"))]

                    tree_vault,

                    path: owned_path,

                    file_manager,

                    chunk_cache: ChunkCache::new(2000, 160_384),

                    threadpool: ThreadPool::new(parallelization),

                    schemas: RwLock::new(configuration.initial_schemas),

                    available_databases: RwLock::default(),

                    open_roots: Mutex::default(),

                    key_value_persistence,

                    check_view_integrity_on_database_open,

                    relay: Relay::default(),

                }),

            })

        })

        .await??;

        storage.cache_available_databases().await?;

        storage.create_admin_database_if_needed().await?;

        Ok(storage)

    }

    pub fn path(&self) -> &Path {

        &self.data.path

    }

    async fn lookup_or_create_id(

        configuration: &StorageConfiguration,

        path: &Path,

    ) -> Result<StorageId, Error> {

        Ok(StorageId(if let Some(id) = configuration.unique_id {

            let id_path = path.join("server-id");

            if id_path.exists() {

                let existing_id = String::from_utf8(

                    File::open(id_path)

                        .and_then(|mut f| async move {

                            let mut bytes = Vec::new();

                            f.read_to_end(&mut bytes).await.map(|_| bytes)

                        })

                        .await

                        .expect("error reading server-id file"),

                )

                .expect("server-id contains invalid data");

                existing_id.parse().expect("server-id isn't numeric")

                let id = { thread_rng().gen::<u64>() };

                File::create(id_path)

                    .and_then(|mut file| async move {

                        let id = id.to_string();

                        file.write_all(id.as_bytes()).await?;

                        file.shutdown().await

                    })

                    .await

                    .map_err(|err| {

                    })?;

                id

    }

    async fn cache_available_databases(&self) -> Result<(), Error> {

        let available_databases = self

            .admin()

            .await

            .view::<ByName>()

            .query()

            .await?

            .into_iter()

            .map(|map| (map.key, map.value))

            .collect();

        let mut storage_databases = fast_async_write!(self.data.available_databases);

        *storage_databases = available_databases;

        Ok(())

    }

    async fn create_admin_database_if_needed(&self) -> Result<(), Error> {

        self.register_schema::<Admin>().await?;

        match self.database::<Admin>(ADMIN_DATABASE_NAME).await {

            Ok(_) => {}

                self.create_database::<Admin>(ADMIN_DATABASE_NAME, true)

                    .await?;

        Ok(())

    }

    pub(crate) fn parallelization(&self) -> usize {

        self.data.parallelization

    }

    pub(crate) fn vault(&self) -> &Arc<Vault> {

        &self.data.vault

    }

    pub(crate) fn tree_vault(&self) -> Option<&TreeVault> {

        self.data.tree_vault.as_ref()

    }

    pub(crate) fn default_encryption_key(&self) -> Option<&KeyId> {

        self.data.default_encryption_key.as_ref()

    }

    pub async fn register_schema<DB: Schema>(&self) -> Result<(), Error> {

        let mut schemas = fast_async_write!(self.data.schemas);

                DB::schema_name(),

                Arc::new(StorageSchemaOpener::<DB>::new()?),

            .is_none()

            Ok(())

    }

    pub(crate) async fn open_roots(&self, name: &str) -> Result<Context, Error> {

        let mut open_roots = fast_async_lock!(self.data.open_roots);

        if let Some(roots) = open_roots.get(name) {

            Ok(roots.clone())

            let task_self = self.clone();

            let task_name = name.to_string();

            let roots = tokio::task::spawn_blocking(move || {

                let mut config = nebari::Config::new(task_self.data.path.join(task_name))

                    .file_manager(task_self.data.file_manager.clone())

                    .cache(task_self.data.chunk_cache.clone())

                    .shared_thread_pool(&task_self.data.threadpool);

                if let Some(vault) = task_self.data.tree_vault.clone() {

                    config = config.vault(vault);

                }

                config.open().map_err(Error::from)

            })

            .await

            .unwrap()?;

            let context = Context::new(roots, self.data.key_value_persistence.clone());

            open_roots.insert(name.to_owned(), context.clone());

            Ok(context)

    }

    pub(crate) fn tasks(&self) -> &'_ TaskManager {

        &self.data.tasks

    }

    pub(crate) fn check_view_integrity_on_database_open(&self) -> bool {

        self.data.check_view_integrity_on_database_open

    }

    pub(crate) fn relay(&self) -> &'_ Relay {

        &self.data.relay

    }

    fn validate_name(name: &str) -> Result<(), Error> {

        if name.chars().enumerate().all(|(index, c)| {

            c.is_ascii_alphanumeric()

                || (index == 0 && c == '_')

                || (index > 0 && (c == '.' || c == '-'))

        }) {

            Ok(())

            Err(Error::Core(bonsaidb_core::Error::InvalidDatabaseName(

                name.to_owned(),

            )))

    }

    pub async fn admin(&self) -> Database {

            self.open_roots(ADMIN_DATABASE_NAME).await.unwrap(),

            self.clone(),

        .unwrap()

    }

    pub(crate) async fn database_without_schema(&self, name: &str) -> Result<Database, Error> {

        let schema = {

            let available_databases = fast_async_read!(self.data.available_databases);

            available_databases

                .get(name)

                .ok_or_else(|| {

                    Error::Core(bonsaidb_core::Error::DatabaseNotFound(name.to_string()))

                })?

                .clone()

        let mut schemas = fast_async_write!(self.data.schemas);

        if let Some(schema) = schemas.get_mut(&schema) {

            let db = schema.open(name.to_string(), self.clone()).await?;

            Ok(db)

    }

    pub async fn database_without_schema_internal(&self, name: &str) -> Result<Database, Error> {

        self.database_without_schema(name).await

    }

    async fn update_user_with_named_id<

        'user,

        'other,

        Col: NamedCollection<PrimaryKey = u64>,

        U: Nameable<'user, u64> + Send + Sync,

        O: Nameable<'other, u64> + Send + Sync,

        F: FnOnce(&mut CollectionDocument<User>, u64) -> bool,

    >(

        &self,

        user: U,

        other: O,

        callback: F,

    ) -> Result<(), bonsaidb_core::Error> {

        let admin = self.admin().await;

        let other = other.name()?;

        let (user, other) =

            futures::try_join!(User::load(user.name()?, &admin), other.id::<Col, _>(&admin),)?;

        match (user, other) {

            (Some(mut user), Some(other)) => {

                if callback(&mut user, other) {

                    user.update(&admin).await?;

                }

                Ok(())

    }

    pub fn new() -> Result<Self, Error> {

        let schematic = DB::schematic()?;

        Ok(Self {

            schematic,

            _phantom: PhantomData::default(),

        })

    }

    async fn open(&self, name: String, storage: Storage) -> Result<Database, Error> {

        let roots = storage.open_roots(&name).await?;

        let db = Database::new::<DB, _>(name, roots, storage).await?;

        Ok(db)

    }

        tracing::instrument(skip(name, schema, only_if_needed))

    ) -> Result<(), bonsaidb_core::Error> {

        Self::validate_name(name)?;

            let schemas = fast_async_read!(self.data.schemas);

            if !schemas.contains_key(&schema) {

                return Err(bonsaidb_core::Error::SchemaNotRegistered(schema));

            }

        let mut available_databases = fast_async_write!(self.data.available_databases);

        let admin = self.admin().await;

        if !admin

            .view::<database::ByName>()

            .with_key(name.to_ascii_lowercase())

            .query()

            .await?

            .is_empty()

            if only_if_needed {

                return Ok(());

            }

            return Err(bonsaidb_core::Error::DatabaseNameAlreadyTaken(

                name.to_string(),

            ));

        }

        admin

            .collection::<DatabaseRecord>()

            .push(&admin::Database {

                name: name.to_string(),

                schema: schema.clone(),

            })

            .await?;

        available_databases.insert(name.to_string(), schema);

        Ok(())

    }

    async fn database<DB: Schema>(

        &self,

        name: &str,

    ) -> Result<Self::Database, bonsaidb_core::Error> {

        let db = self.database_without_schema(name).await?;

        if db.data.schema.name == DB::schema_name() {

            Ok(db)

    }

    #[cfg_attr(feature = "tracing", tracing::instrument(skip(name)))]

    async fn delete_database(&self, name: &str) -> Result<(), bonsaidb_core::Error> {

        let admin = self.admin().await;

        let mut available_databases = fast_async_write!(self.data.available_databases);

        available_databases.remove(name);

        let mut open_roots = fast_async_lock!(self.data.open_roots);

        open_roots.remove(name);

        let database_folder = self.path().join(name);

        if database_folder.exists() {

            let file_manager = self.data.file_manager.clone();

            tokio::task::spawn_blocking(move || file_manager.delete_directory(&database_folder))

                .await

                .unwrap()

                .map_err(Error::Nebari)?;

        }

        if let Some(entry) = admin

            .view::<database::ByName>()

            .with_key(name.to_ascii_lowercase())

            .query()

            .await?

            .first()

            admin.delete::<DatabaseRecord, _>(&entry.source).await?;

            Ok(())

            return Err(bonsaidb_core::Error::DatabaseNotFound(name.to_string()));

    }

    #[cfg_attr(feature = "tracing", tracing::instrument)]

    async fn list_databases(&self) -> Result<Vec<connection::Database>, bonsaidb_core::Error> {

        let available_databases = fast_async_read!(self.data.available_databases);

        Ok(available_databases

            .iter()

            .map(|(name, schema)| connection::Database {

                name: name.to_string(),

                schema: schema.clone(),

            })

            .collect())

    }

    #[cfg_attr(feature = "tracing", tracing::instrument)]

    async fn list_available_schemas(&self) -> Result<Vec<SchemaName>, bonsaidb_core::Error> {

        let available_databases = fast_async_read!(self.data.available_databases);

        Ok(available_databases.values().unique().cloned().collect())

    }

    #[cfg_attr(feature = "tracing", tracing::instrument(skip(username)))]

    async fn create_user(&self, username: &str) -> Result<u64, bonsaidb_core::Error> {

        let result = self

            .admin()

            .collection::<User>()

            .push(&User::default_with_username(username))

            .await?;

        Ok(result.id)

    }

    #[cfg_attr(feature = "tracing", tracing::instrument(skip(user)))]

    ) -> Result<(), bonsaidb_core::Error> {

        let admin = self.admin().await;

        let doc = User::load(user, &admin)

            .await?

            .ok_or(bonsaidb_core::Error::UserNotFound)?;

        doc.delete(&admin).await?;

        Ok(())

    }

    #[cfg_attr(feature = "tracing", tracing::instrument(skip(user, password)))]

    ) -> Result<(), bonsaidb_core::Error> {

        let admin = self.admin().await;

        let mut user = User::load(user, &admin)

            .await?

            .ok_or(bonsaidb_core::Error::UserNotFound)?;

        user.contents.argon_hash = Some(self.data.argon.hash(user.header.id, password).await?);

        user.update(&admin).await

    }

    #[cfg_attr(feature = "tracing", tracing::instrument(skip(user)))]

    ) -> Result<Authenticated, bonsaidb_core::Error> {

        let admin = self.admin().await;

        let user = User::load(user, &admin)

            .await?

            .ok_or(bonsaidb_core::Error::InvalidCredentials)?;

        match authentication {

            Authentication::Password(password) => {

                let saved_hash = user

                    .contents

                    .argon_hash

                    .clone()

                    .ok_or(bonsaidb_core::Error::InvalidCredentials)?;

                self.data

                    .argon

                    .verify(user.header.id, password, saved_hash)

                    .await?;

                let permissions = user.contents.effective_permissions(&admin).await?;

                Ok(Authenticated {

                    user_id: user.header.id,

                    permissions,

                })

    }

    #[cfg_attr(feature = "tracing", tracing::instrument(skip(user, permission_group)))]

    ) -> Result<(), bonsaidb_core::Error> {

        self.update_user_with_named_id::<PermissionGroup, _, _, _>(

            user,

            permission_group,

            |user, permission_group_id| {

                if user.contents.groups.contains(&permission_group_id) {

                    false

                    user.contents.groups.push(permission_group_id);

                    true

            },

        )

        .await

    }

    #[cfg_attr(feature = "tracing", tracing::instrument(skip(user, permission_group)))]

    ) -> Result<(), bonsaidb_core::Error> {

        self.update_user_with_named_id::<PermissionGroup, _, _, _>(

            user,

            permission_group,

            |user, permission_group_id| {

                let old_len = user.contents.groups.len();

                user.contents.groups.retain(|id| id != &permission_group_id);

                old_len != user.contents.groups.len()

            },

        )

        .await

    }

    #[cfg_attr(feature = "tracing", tracing::instrument(skip(user, role)))]

    ) -> Result<(), bonsaidb_core::Error> {

        self.update_user_with_named_id::<PermissionGroup, _, _, _>(user, role, |user, role_id| {

            if user.contents.roles.contains(&role_id) {

                false

                user.contents.roles.push(role_id);

                true

        })

        .await

    }

    #[cfg_attr(feature = "tracing", tracing::instrument(skip(user, role)))]

    ) -> Result<(), bonsaidb_core::Error> {

        self.update_user_with_named_id::<Role, _, _, _>(user, role, |user, role_id| {

            let old_len = user.contents.roles.len();

            user.contents.roles.retain(|id| id != &role_id);

            old_len != user.contents.roles.len()

        })

        .await

    }

#[test]

fn name_validation_tests() {

    assert!(matches!(Storage::validate_name("azAZ09.-"), Ok(())));

    assert!(matches!(

        Storage::validate_name("_internal-names-work"),

    assert!(matches!(

        Storage::validate_name("-alphaunmericfirstrequired"),

    assert!(matches!(

        Storage::validate_name("\u{2661}"),

}

    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {

        // let formatted_length = format!();

        write!(f, "{:016x}", self.0)

    }

    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {

        Debug::fmt(self, f)

    }

#[derive(Debug, Clone)]

    pub(crate) fn new_if_needed(

        key: Option<KeyId>,

        vault: &Arc<Vault>,

        compression: Option<Compression>,

    ) -> Option<Self> {

        if key.is_none() && compression.is_none() {

            None

            Some(Self {

                key,

                compression,

                vault: vault.clone(),

            })

    }

    fn header(&self, compressed: bool) -> u8 {

        let mut bits = if self.key.is_some() { 0b1000_0000 } else { 0 };

        if compressed {

            if let Some(compression) = self.compression {

                bits |= compression as u8;

            }

        }

        bits

    }

    fn encrypt(&self, payload: &[u8]) -> Result<Vec<u8>, Error> {

        use std::borrow::Cow;

        // TODO this allocates too much. The vault should be able to do an

        // in-place encryption operation so that we can use a single buffer.

        let mut includes_compression = false;

        let compressed = match (payload.len(), self.compression) {

            (128..=usize::MAX, Some(Compression::Lz4)) => {

                includes_compression = true;

                Cow::Owned(lz4_flex::block::compress_prepend_size(payload))

            _ => Cow::Borrowed(payload),

        let mut complete = if let Some(key) = &self.key {

            self.vault.encrypt_payload(key, &compressed, None)?

            compressed.into_owned()

        let header = self.header(includes_compression);

        if header != 0 {

            let header = [b't', b'r', b'v', header];

            complete.splice(0..0, header);

        }

        Ok(complete)

    }

    fn decrypt(&self, payload: &[u8]) -> Result<Vec<u8>, Error> {

        use std::borrow::Cow;

        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;

            let decrypted = if encrypted {

                Cow::Owned(self.vault.decrypt_payload(payload, None)?)

                Cow::Borrowed(payload)

            return Ok(match Compression::from_u8(compression) {

                    lz4_flex::block::decompress_size_prepended(&decrypted).map_err(Error::from)?

                None => decrypted.into_owned(),

        }

        self.vault.decrypt_payload(payload, None)

    }