#[derive(Debug, Clone)]

#[derive(Debug, Default)]

    pub fn unregister(&mut self, subscriber_id: u64) {

        if let Some(session_id) = self

            .subscribers

            .remove(&subscriber_id)

            .and_then(|sub| sub.session_id)

        }

    }

    fn drop(&mut self) {

        let mut session = self.session.lock();

        if let Some(id) = session.id.take() {

            if let Some(storage) = self.storage.upgrade() {

                let mut sessions = storage.sessions.write();

                sessions.sessions.remove(&id);

                // Remove all subscribers.

                let mut sessions = storage.subscribers.write();

                for id in sessions

                    .subscribers_by_session

                    .remove(&id)

                    .into_iter()

                    .flatten()

            }

    }

#[derive(Debug, Default)]

#[derive(Debug, Clone)]

    fn from(instance: StorageInstance) -> Self {

        Self {

            instance,

            authentication: None,

            effective_session: None,

        }

    }

    pub 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)?;

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

        let vault = {

            let vault_key_storage = match configuration.vault_key_storage {

                Some(storage) => storage,

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

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

            Arc::new(Vault::initialize(

                storage_lock.id(),

                &owned_path,

                vault_key_storage,

            )?)

        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,

        );

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

        let tree_vault = TreeVault::new_if_needed(default_encryption_key.clone(), &vault);

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

        let tree_vault = TreeVault::new_if_needed(configuration.default_compression);

        let authenticated_permissions = configuration.authenticated_permissions;

        let storage = Self {

            instance: StorageInstance {

                data: Arc::new(Data {

                    lock: storage_lock,

                    tasks,

                    parallelization,

                    subscribers: Arc::default(),

                    authenticated_permissions,

                    sessions: RwLock::default(),

                    #[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(),

                }),

            },

            authentication: None,

            effective_session: None,

        };

        storage.cache_available_databases()?;

        storage.create_admin_database_if_needed()?;

        Ok(storage)

    }

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

        let name = name.to_owned();

        self.instance

            .database_without_schema(&name, Some(self), None)

    }

    fn lookup_or_create_id(

        configuration: &StorageConfiguration,

        path: &Path,

    ) -> Result<StorageLock, Error> {

        let id_path = {

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

            if storage_id.exists() {

                storage_id

                path.join("storage-id")

        let (id, file) = if let Some(id) = configuration.unique_id {

            if id_path.exists() {

                let mut file = File::open(id_path)?;

                file.lock_exclusive()?;

                let mut bytes = Vec::new();

                file.read_to_end(&mut bytes)?;

                let existing_id =

                    String::from_utf8(bytes).expect("server-id contains invalid data");

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

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

                let mut file = File::create(id_path)?;

                file.lock_exclusive()?;

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

                (id, file)

        Ok(StorageLock::new(StorageId(id), file))

    }

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

        let available_databases = self

            .admin()

            .view::<ByName>()

            .query()?

            .into_iter()

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

            .collect();

        let mut storage_databases = self.instance.data.available_databases.write();

        *storage_databases = available_databases;

        Ok(())

    }

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

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

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

            Ok(_) => {}

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

        Ok(())

    }

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

        self.instance.data.parallelization

    }

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

        &self.instance.data.vault

    }

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

        self.instance.data.tree_vault.as_ref()

    }

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

        self.instance.data.default_encryption_key.as_ref()

    }

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

        let mut schemas = self.instance.data.schemas.write();

        if schemas

            .insert(

                DB::schema_name(),

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

            .is_none()

            Ok(())

    }

    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 fn into_async(self) -> crate::AsyncStorage {

        self.into_async_with_runtime(tokio::runtime::Handle::current())

    }

    pub fn into_async_with_runtime(self, runtime: tokio::runtime::Handle) -> crate::AsyncStorage {

        crate::AsyncStorage {

            storage: self,

            runtime: Arc::new(runtime),

        }

    }

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

        let mut open_roots = self.data.open_roots.lock();

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

            Ok(roots.clone())

            let task_name = name.to_string();

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

                .file_manager(self.data.file_manager.clone())

                .cache(self.data.chunk_cache.clone())

                .shared_thread_pool(&self.data.threadpool);

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

                config = config.vault(vault);

            }

            let roots = config.open().map_err(Error::from)?;

            let context = Context::new(

                roots,

                self.data.key_value_persistence.clone(),

                Some(self.data.lock.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

    }

    pub(crate) fn database_without_schema(

        &self,

        name: &str,

        storage: Option<&Storage>,

        expected_schema: Option<SchemaName>,

    ) -> Result<Database, Error> {

        let stored_schema = {

            let available_databases = self.data.available_databases.read();

            available_databases

                .get(name)

                .ok_or_else(|| {

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

                })?

                .clone()

        if let Some(expected_schema) = expected_schema {

            if stored_schema != expected_schema {

            }

        }

        let mut schemas = self.data.schemas.write();

        let storage =

            storage.map_or_else(|| Cow::Owned(Storage::from(self.clone())), Cow::Borrowed);

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

            let db = schema.open(name.to_string(), storage.as_ref())?;

            Ok(db)

    }

    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) -> Result<bool, bonsaidb_core::Error>,

    >(

        &self,

        user: U,

        other: O,

        callback: F,

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

        let admin = self.admin();

        let other = other.name()?;

        let user = User::load(user.name()?, &admin)?;

        let other = other.id::<Col, _>(&admin)?;

        match (user, other) {

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

                if callback(&mut user, other)? {

                    user.update(&admin)?;

                }

                Ok(())

    }

    fn authenticate_inner(

        &self,

        authentication: bonsaidb_core::connection::Authentication,

        loaded_user: Option<CollectionDocument<User>>,

        current_session_id: Option<SessionId>,

        admin: &Database,

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

        use bonsaidb_core::connection::Authentication;

        match authentication {

                id,

                now,

                now_hash,

                algorithm,

            } => self.begin_token_authentication(id, now, &now_hash, algorithm, admin),

            Authentication::TokenChallengeResponse(hash) => {

                let session_id =

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

                self.finish_token_authentication(session_id, &hash, admin)

            Authentication::Password { user, password } => {

                let user = match loaded_user {

                    Some(user) => user,

                let saved_hash = user

                    .contents

                    .argon_hash

                    .clone()

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

                self.data

                    .argon

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

                self.assume_user(user, admin)

    }

    fn assume_user(

        &self,

        user: CollectionDocument<User>,

        admin: &Database,

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

        let permissions = user.contents.effective_permissions(

            admin,

            &admin.storage().instance.data.authenticated_permissions,

        )?;

        let mut sessions = self.data.sessions.write();

        sessions.last_session_id += 1;

        let session_id = SessionId(sessions.last_session_id);

        let session = Session {

            id: Some(session_id),

            authentication: SessionAuthentication::Identity(Arc::new(Identity::User {

                id: user.header.id,

                username: user.contents.username,

            })),

            permissions,

        };

        let authentication = Arc::new(AuthenticatedSession {

            storage: Arc::downgrade(&self.data),

            session: Mutex::new(session.clone()),

        });

        sessions.sessions.insert(session_id, authentication.clone());

        Ok(Storage {

            instance: self.clone(),

            authentication: Some(authentication),

            effective_session: Some(Arc::new(session)),

        })

    }

    fn assume_role(

        &self,

        role: CollectionDocument<Role>,

        admin: &Database,

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

        let permissions = role.contents.effective_permissions(

            admin,

            &admin.storage().instance.data.authenticated_permissions,

        )?;

        let mut sessions = self.data.sessions.write();

        sessions.last_session_id += 1;

        let session_id = SessionId(sessions.last_session_id);

        let session = Session {

            id: Some(session_id),

            authentication: SessionAuthentication::Identity(Arc::new(Identity::Role {

                id: role.header.id,

                name: role.contents.name,

            })),

            permissions,

        };

        let authentication = Arc::new(AuthenticatedSession {

            storage: Arc::downgrade(&self.data),

            session: Mutex::new(session.clone()),

        });

        sessions.sessions.insert(session_id, authentication.clone());

        Ok(Storage {

            instance: self.clone(),

            authentication: Some(authentication),

            effective_session: Some(Arc::new(session)),

        })

    }

    fn add_permission_group_to_user_inner(

        user: &mut CollectionDocument<User>,

        permission_group_id: u64,

    ) -> bool {

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

            false

            user.contents.groups.push(permission_group_id);

            true

    }

    fn remove_permission_group_from_user_inner(

        user: &mut CollectionDocument<User>,

        permission_group_id: u64,

    ) -> bool {

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

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

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

    }

    fn add_role_to_user_inner(user: &mut CollectionDocument<User>, role_id: u64) -> bool {

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

            false

            user.contents.roles.push(role_id);

            true

    }

    fn remove_role_from_user_inner(user: &mut CollectionDocument<User>, role_id: u64) -> bool {

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

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

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

    }

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

        let schematic = DB::schematic()?;

        Ok(Self {

            schematic,

            _phantom: PhantomData,

        })

    }

    fn schematic(&self) -> &'_ Schematic {

        &self.schematic

    }

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

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

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

        Ok(db)

    }

    fn admin(&self) -> Self::Database {

        Database::new::<Admin, _>(

            ADMIN_DATABASE_NAME,

            self.open_roots(ADMIN_DATABASE_NAME).unwrap(),

            &Storage::from(self.clone()),

        )

        .unwrap()

    }

    fn create_database_with_schema(

        &self,

        name: &str,

        schema: SchemaName,

        only_if_needed: bool,

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

        Storage::validate_name(name)?;

            let schemas = self.data.schemas.read();

            if !schemas.contains_key(&schema) {

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

            }

        }

        let mut available_databases = self.data.available_databases.write();

        let admin = self.admin();

        if !available_databases.contains_key(name) {

            admin

                .collection::<DatabaseRecord>()

                .push(&admin::Database {

                    name: name.to_string(),

                    schema: schema.clone(),

                })?;

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

        } else if !only_if_needed {

            return Err(bonsaidb_core::Error::DatabaseNameAlreadyTaken(

                name.to_string(),

            ));

        }

        Ok(())

    }

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

        self.database_without_schema(name, None, Some(DB::schema_name()))

            .map_err(bonsaidb_core::Error::from)

    }

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

        let admin = self.admin();

        let mut available_databases = self.data.available_databases.write();

        available_databases.remove(name);

        let mut open_roots = self.data.open_roots.lock();

        open_roots.remove(name);

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

        if database_folder.exists() {

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

            file_manager

                .delete_directory(&database_folder)

                .map_err(Error::Nebari)?;

        }

        if let Some(entry) = admin

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

            .with_key(name)

            .query()?

            .first()

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

            Ok(())

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

    }

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

        let available_databases = self.data.available_databases.read();

        Ok(available_databases

            .iter()

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

                name: name.to_string(),

                schema: schema.clone(),

            })

            .collect())

    }

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

        let available_databases = self.data.available_databases.read();

        let schemas = self.data.schemas.read();

        Ok(available_databases

            .values()

            .unique()

            .filter_map(|name| {

                schemas

                    .get(name)

                    .map(|opener| SchemaSummary::from(opener.schematic()))

            })

            .collect())

    }

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

        let result = self

            .admin()

            .collection::<User>()

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

        Ok(result.id)

    }

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

        &self,

        user: U,

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

        let admin = self.admin();

        let user = User::load(user, &admin)?.ok_or(bonsaidb_core::Error::UserNotFound)?;

        user.delete(&admin)?;

        Ok(())

    }

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

        &self,

        user: U,

        password: bonsaidb_core::connection::SensitiveString,

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

        let admin = self.admin();

        let mut user = User::load(user, &admin)?.ok_or(bonsaidb_core::Error::UserNotFound)?;

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

        user.update(&admin)

    }

    fn session(&self) -> Option<&Session> {

        self.effective_session.as_deref()

    }

    fn admin(&self) -> Self::Database {

        self.instance.admin()

    }

    fn create_database_with_schema(

        &self,

        name: &str,

        schema: SchemaName,

        only_if_needed: bool,

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

        self.check_permission(

            database_resource_name(name),

            &BonsaiAction::Server(ServerAction::CreateDatabase),

        )?;

        self.instance

            .create_database_with_schema(name, schema, only_if_needed)

    }

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

        self.instance.database::<DB>(name)

    }

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

        self.check_permission(

            database_resource_name(name),

            &BonsaiAction::Server(ServerAction::DeleteDatabase),

        )?;

        self.instance.delete_database(name)

    }

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

        self.check_permission(

            bonsaidb_resource_name(),

            &BonsaiAction::Server(ServerAction::ListDatabases),

        )?;

        self.instance.list_databases()

    }

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

        self.check_permission(

            bonsaidb_resource_name(),

            &BonsaiAction::Server(ServerAction::ListAvailableSchemas),

        )?;

        self.instance.list_available_schemas()

    }

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

        self.check_permission(

            bonsaidb_resource_name(),

            &BonsaiAction::Server(ServerAction::CreateUser),

        )?;

        self.instance.create_user(username)

    }

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

        &self,

        user: U,

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

        let admin = self.admin();

        let user = user.name()?;

        let user_id = user

            .id::<User, _>(&admin)?

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

        self.check_permission(

            user_resource_name(user_id),

            &BonsaiAction::Server(ServerAction::DeleteUser),

        )?;

        self.instance.delete_user(user)

    }

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

        &self,

        user: U,

        password: bonsaidb_core::connection::SensitiveString,

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

        let admin = self.admin();

        let user = user.name()?;

        let user_id = user

            .id::<User, _>(&admin)?

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

        self.check_permission(

            user_resource_name(user_id),

            &BonsaiAction::Server(ServerAction::SetPassword),

        )?;

        self.instance.set_user_password(user, password)

    }

    fn authenticate(

        &self,

        authentication: bonsaidb_core::connection::Authentication,

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

        let admin = self.admin();

        let mut loaded_user = None;

        match &authentication {

            bonsaidb_core::connection::Authentication::Token { id, .. } => {

                self.check_permission(

                    bonsaidb_core::permissions::bonsai::authentication_token_resource_name(*id),

                    &BonsaiAction::Server(ServerAction::Authenticate(

                        bonsaidb_core::connection::AuthenticationMethod::Token,