#[derive_where(Clone)]

#[derive(Default)]

    fn deref(&self) -> &Self::Target {

        &self.0

    }

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

        let request_processor = Manager::default();

        for _ in 0..configuration.request_workers {

            request_processor.spawn_worker();

        }

        let storage = Storage::open(configuration.storage.with_schema::<Hosted>()?).await?;

        storage.create_database::<Hosted>("_hosted", true).await?;

        let default_permissions = Permissions::from(configuration.default_permissions);

        let authenticated_permissions = Permissions::from(configuration.authenticated_permissions);

        let server = Self {

            data: Arc::new(Data {

                clients: RwLock::default(),

                storage,

                endpoint: RwLock::default(),

                request_processor,

                default_permissions,

                authenticated_permissions,

                client_simultaneous_request_limit: configuration.client_simultaneous_request_limit,

                primary_tls_key: CachedCertifiedKey::default(),

                primary_domain: configuration.server_name,

                #[cfg(feature = "acme")]

                acme: configuration.acme,

                #[cfg(feature = "acme")]

                alpn_keys: AlpnKeys::default(),

                shutdown: Shutdown::new(),

                relay: Relay::default(),

                subscribers: Arc::default(),

                _backend: PhantomData::default(),

            }),

        };

        B::initialize(&server).await;

        Ok(server)

    }

    pub fn pinned_certificate_path(&self) -> PathBuf {

        self.path().join("pinned-certificate.der")

    }

    pub fn primary_domain(&self) -> &str {

        &self.data.primary_domain

    }

    pub async fn admin(&self) -> ServerDatabase<B> {

        let db = self.data.storage.admin().await;

        ServerDatabase {

            server: self.clone(),

            db,

        }

    }

    pub(crate) async fn hosted(&self) -> ServerDatabase<B> {

        let db = self

            .data

            .storage

            .database::<Hosted>("_hosted")

            .await

            .unwrap();

        ServerDatabase {

            server: self.clone(),

            db,

        }

    }

    pub async fn install_self_signed_certificate(&self, overwrite: bool) -> Result<(), Error> {

        let keypair = KeyPair::new_self_signed(&self.data.primary_domain);

        if self.certificate_chain().await.is_ok() && !overwrite {

            return Err(Error::Core(bonsaidb_core::Error::Configuration(String::from("Certificate already installed. Enable overwrite if you wish to replace the existing certificate."))));

        }

        self.install_certificate(keypair.certificate_chain(), keypair.private_key())

            .await?;

        tokio::fs::write(

            self.pinned_certificate_path(),

            keypair.end_entity_certificate().as_ref(),

        )

        .await?;

        Ok(())

    }

    pub async fn install_certificate(

        &self,

        certificate_chain: &CertificateChain,

        private_key: &PrivateKey,

    ) -> Result<(), Error> {

        let db = self.hosted().await;

        TlsCertificate::entry(&self.data.primary_domain, &db)

            .update_with(|cert: &mut TlsCertificate| {

                cert.certificate_chain = certificate_chain.clone();

                cert.private_key = SerializablePrivateKey(private_key.clone());

            })

            .or_insert_with(|| TlsCertificate {

                domains: vec![self.data.primary_domain.clone()],

                private_key: SerializablePrivateKey(private_key.clone()),

                certificate_chain: certificate_chain.clone(),

            })

            .await?;

        self.refresh_certified_key().await?;

        let pinned_certificate_path = self.pinned_certificate_path();

        if pinned_certificate_path.exists() {

            tokio::fs::remove_file(&pinned_certificate_path).await?;

        }

        Ok(())

    }

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

        let certificate = self.tls_certificate().await?;

        let mut cached_key = self.data.primary_tls_key.lock();

        let private_key = rustls::PrivateKey(

            fabruic::dangerous::PrivateKey::as_ref(&certificate.private_key.0).to_vec(),

        );

        let private_key = rustls::sign::any_ecdsa_type(&Arc::new(private_key))?;

        let certificates = certificate

            .certificate_chain

            .iter()

            .map(|cert| rustls::Certificate(cert.as_ref().to_vec()))

            .collect::<Vec<_>>();

        let certified_key = Arc::new(CertifiedKey::new(certificates, private_key));

        *cached_key = Some(certified_key);

        Ok(())

    }

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

        let db = self.hosted().await;

        let (_, certificate) = db

            .view::<TlsCertificatesByDomain>()

            .with_key(self.data.primary_domain.clone())

            .query_with_collection_docs()

            .await?

            .into_iter()

            .next()

            .ok_or_else(|| {

            })?;

        Ok(certificate.contents)

    }

    pub async fn certificate_chain(&self) -> Result<CertificateChain, Error> {

        let db = self.hosted().await;

        if let Some(mapping) = db

            .view::<TlsCertificatesByDomain>()

            .with_key(self.data.primary_domain.clone())

            .query()

            .await?

            .into_iter()

            .next()

            Ok(mapping.value)

            Err(Error::Core(bonsaidb_core::Error::Configuration(format!(

                "no certificate found for {}",

                self.data.primary_domain

            ))))

    }

    pub async fn listen_on(&self, port: u16) -> Result<(), Error> {

        let certificate = self.tls_certificate().await?;

        let keypair =

            KeyPair::from_parts(certificate.certificate_chain, certificate.private_key.0)?;

        let mut builder = Endpoint::builder();

        builder.set_protocols([CURRENT_PROTOCOL_VERSION.as_bytes().to_vec()]);

        builder.set_address(([0; 8], port).into());

        builder

            .set_max_idle_timeout(None)

            .map_err(|err| Error::Core(bonsaidb_core::Error::Transport(err.to_string())))?;

        builder.set_server_key_pair(Some(keypair));

        let mut server = builder

            .build()

            .map_err(|err| Error::Core(bonsaidb_core::Error::Transport(err.to_string())))?;

        {

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

            *endpoint = Some(server.clone());

        let mut shutdown_watcher = self

            .data

            .shutdown

            .watcher()

            .expect("server already shut down");

        while let Some(result) = tokio::select! {

            shutdown_state = shutdown_watcher.wait_for_shutdown() => {

            msg = server.next() => msg

        } {

            let connection = result.accept::<()>().await?;

            let task_self = self.clone();

            tokio::spawn(async move {

                let address = connection.remote_address();

                if let Err(err) = task_self.handle_bonsai_connection(connection).await {

                }

            });

        Ok(())

    }

    async fn initialize_client(

        &self,

        transport: Transport,

        address: SocketAddr,

        sender: Sender<CustomApiResult<B::CustomApi>>,

    ) -> Option<OwnedClient<B>> {

        if !self.data.default_permissions.allowed_to(

            &bonsaidb_resource_name(),

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

        ) {

        }

        let client = loop {

            let next_id = CONNECTED_CLIENT_ID_COUNTER.fetch_add(1, Ordering::SeqCst);

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

            if let hash_map::Entry::Vacant(e) = clients.entry(next_id) {

                let client = OwnedClient::new(next_id, address, transport, sender, self.clone());

                e.insert(client.clone());

                break client;

            B::client_connected(&client, self).await,

            Some(client)

    }

    async fn disconnect_client(&self, id: u32) {

        if let Some(client) = {

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

            clients.remove(&id)

            B::client_disconnected(client, self).await;

    }

    async fn handle_bonsai_connection(

        &self,

        mut connection: fabruic::Connection<()>,

    ) -> Result<(), Error> {

        if let Some(incoming) = connection.next().await {

            let incoming = match incoming {

                Ok(incoming) => incoming,

            match incoming

            .accept::<networking::Payload<Response<CustomApiResult<B::CustomApi>>>, networking::Payload<Request<<B::CustomApi as CustomApi>::Request>>>()

                Ok((sender, receiver)) => {

                    let (api_response_sender, api_response_receiver) = flume::unbounded();

                    if let Some(disconnector) = self

                        .initialize_client(

                            Transport::Bonsai,

                            connection.remote_address(),

                            api_response_sender,

                        )

                    {

                        let task_sender = sender.clone();

                        tokio::spawn(async move {

                            while let Ok(response) = api_response_receiver.recv_async().await {

                            let _ = connection.close().await;

                        });

                        let task_self = self.clone();

                        tokio::spawn(async move {

                            if let Err(err) = task_self

                                .handle_stream(disconnector, sender, receiver)

                                .await

                            }

                        });

                    } else {

        Ok(())

    }

    async fn handle_client_requests(

        &self,

        client: ConnectedClient<B>,

        request_receiver: flume::Receiver<Payload<Request<<B::CustomApi as CustomApi>::Request>>>,

        response_sender: flume::Sender<Payload<Response<CustomApiResult<B::CustomApi>>>>,

    ) {

        let notify = Arc::new(Notify::new());

        let requests_in_queue = Arc::new(AtomicUsize::new(0));

        loop {

            let current_requests = requests_in_queue.load(Ordering::SeqCst);

            if current_requests == self.data.client_simultaneous_request_limit {

                notify.notified().await;

            } else if requests_in_queue

                .compare_exchange(

                    current_requests,

                    current_requests + 1,

                    Ordering::SeqCst,

                    Ordering::SeqCst,

                )

                .is_ok()

                let payload = match request_receiver.recv_async().await {

                    Ok(payload) => payload,

                    Err(_) => break,

                let id = payload.id;

                let task_sender = response_sender.clone();

                let notify = notify.clone();

                let requests_in_queue = requests_in_queue.clone();

                self.handle_request_through_worker(

                    payload.wrapped,

                    move |response| async move {

                        drop(task_sender.send(Payload {

                            id,

                            wrapped: response,

                        }));

                        requests_in_queue.fetch_sub(1, Ordering::SeqCst);

                        notify.notify_one();

                        Ok(())

                    },

                    client.clone(),

                    self.data.subscribers.clone(),

                    response_sender.clone(),

                )

                .await

                .unwrap();

            }

    }

    async fn handle_request_through_worker<

        F: FnOnce(Response<CustomApiResult<B::CustomApi>>) -> R + Send + 'static,

        R: Future<Output = Result<(), Error>> + Send,

    >(

        &self,

        request: Request<<B::CustomApi as CustomApi>::Request>,

        callback: F,

        client: ConnectedClient<B>,

        subscribers: Arc<RwLock<HashMap<u64, Subscriber>>>,

        response_sender: flume::Sender<Payload<Response<CustomApiResult<B::CustomApi>>>>,

    ) -> Result<(), Error> {

        let job = self

            .data

            .request_processor

            .enqueue(ClientRequest::<B>::new(

                request,

                self.clone(),

                client,

                subscribers,

                response_sender,

            ))

            .await;

        tokio::spawn(async move {

            let result = job.receive().await?;

            let result =

                result.unwrap_or_else(|err| Response::Error(Arc::try_unwrap(err).unwrap().into()));

            callback(result).await?;

            Result::<(), Error>::Ok(())

        });

        Ok(())

    }

    async fn handle_stream(

        &self,

        client: OwnedClient<B>,

        sender: fabruic::Sender<Payload<Response<CustomApiResult<B::CustomApi>>>>,

        mut receiver: fabruic::Receiver<Payload<Request<<B::CustomApi as CustomApi>::Request>>>,

    ) -> Result<(), Error> {

        let (payload_sender, payload_receiver) = flume::unbounded();

        tokio::spawn(async move {

            while let Ok(payload) = payload_receiver.recv_async().await {

                if sender.send(&payload).is_err() {

                }

        });

        let (request_sender, request_receiver) =

            flume::bounded::<Payload<Request<<B::CustomApi as CustomApi>::Request>>>(

                self.data.client_simultaneous_request_limit,

            );

        let task_self = self.clone();

        tokio::spawn(async move {

            task_self

                .handle_client_requests(client.clone(), request_receiver, payload_sender)

                .await;

        });

        while let Some(payload) = receiver.next().await {

            drop(request_sender.send_async(payload?).await);

        Ok(())

    }

    async fn forward_notifications_for(

        &self,

        subscriber_id: u64,

        receiver: flume::Receiver<Arc<Message>>,

        sender: flume::Sender<Payload<Response<CustomApiResult<B::CustomApi>>>>,

    ) {

        while let Ok(message) = receiver.recv_async().await {

            if sender

                .send(Payload {

                    id: None,

                    wrapped: Response::Database(DatabaseResponse::MessageReceived {

                        subscriber_id,

                        topic: message.topic.clone(),

                        payload: Bytes::from(&message.payload[..]),

                    }),

                })

                .is_err()

            }

    pub async fn shutdown(&self, timeout: Option<Duration>) -> Result<(), Error> {

        if let Some(timeout) = timeout {

            self.data.shutdown.graceful_shutdown(timeout).await;

            self.data.shutdown.shutdown().await;

        Ok(())

    }

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

        const GRACEFUL_SHUTDOWN: usize = 1;

        const TERMINATE: usize = 2;

        enum SignalShutdownState {

            Running,

            ShuttingDown(flume::Receiver<()>),

        }

        let shutdown_state = Arc::new(Mutex::new(SignalShutdownState::Running));

        let flag = Arc::new(AtomicUsize::default());

        let register_hook = |flag: &Arc<AtomicUsize>| {

            signal_hook::flag::register_usize(SIGINT, flag.clone(), GRACEFUL_SHUTDOWN)?;

            signal_hook::flag::register_usize(SIGTERM, flag.clone(), TERMINATE)?;

            signal_hook::flag::register_usize(SIGQUIT, flag.clone(), TERMINATE)?;

            Result::<(), std::io::Error>::Ok(())

        };

        if let Err(err) = register_hook(&flag) {

                match flag.load(Ordering::Relaxed) {

                    0 => {

                        // No signal

                    }

                let state = fast_async_lock!(shutdown_state);

                if let SignalShutdownState::ShuttingDown(receiver) = &*state {

                }

                tokio::time::sleep(Duration::from_millis(300)).await;

    async fn publish_message(&self, database: &str, topic: &str, payload: Vec<u8>) {

        self.data

            .relay

            .publish_message(Message {

                topic: database_topic(database, topic),

                payload,

            })

    }

    async fn publish_serialized_to_all(&self, database: &str, topics: &[String], payload: Vec<u8>) {

        self.data

            .relay

            .publish_serialized_to_all(

                topics

                    .iter()

                    .map(|topic| database_topic(database, topic))

                    .collect(),

                payload,

            )

    }

    async fn create_subscriber(&self, database: String) -> ServerSubscriber<B> {

        let subscriber = self.data.relay.create_subscriber().await;

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

        let subscriber_id = subscriber.id();

        let receiver = subscriber.receiver().clone();

        subscribers.insert(subscriber_id, subscriber);

        ServerSubscriber {

            server: self.clone(),

            database,

            receiver,

            id: subscriber_id,

        }

    }

    async fn subscribe_to<S: Into<String> + Send>(

        &self,

        subscriber_id: u64,

        database: &str,

        topic: S,

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

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

        if let Some(subscriber) = subscribers.get(&subscriber_id) {

            subscriber

                .subscribe_to(database_topic(database, &topic.into()))

            Ok(())

    }

    async fn unsubscribe_from(

        &self,

        subscriber_id: u64,

        database: &str,

        topic: &str,

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

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

        if let Some(subscriber) = subscribers.get(&subscriber_id) {

            subscriber

                .unsubscribe_from(&database_topic(database, topic))

            Ok(())

    }

    fn deref(&self) -> &Self::Target {

        &self.data.storage

    }

    pub fn new(

        request: Request<<B::CustomApi as CustomApi>::Request>,

        server: CustomServer<B>,

        client: ConnectedClient<B>,

        subscribers: Arc<RwLock<HashMap<u64, Subscriber>>>,

        sender: flume::Sender<Payload<Response<CustomApiResult<B::CustomApi>>>>,

    ) -> Self {

        Self {

            request: Some(request),

            server,

            client,

            subscribers,

            sender,

        }

    }

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

    async fn execute(&mut self) -> Result<Self::Output, Self::Error> {

        let request = self.request.take().unwrap();

        ServerDispatcher {

            server: &self.server,

            client: &self.client,

            subscribers: &self.subscribers,

            response_sender: &self.sender,

        }

        .dispatch(&self.client.permissions().await, request)

        .await

    }

    async fn create_database_with_schema(

        &self,

        name: &str,

        schema: SchemaName,

        only_if_needed: bool,

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

        self.data

            .storage

            .create_database_with_schema(name, schema, only_if_needed)

            .await

    }

    async fn database<DB: Schema>(

        &self,

        name: &str,

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

        let db = self.data.storage.database::<DB>(name).await?;

        Ok(ServerDatabase {

            server: self.clone(),

            db,

        })

    }

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

        self.data.storage.delete_database(name).await

    }

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

        self.data.storage.list_databases().await

    }

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

        self.data.storage.list_available_schemas().await

    }

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

        self.data.storage.create_user(username).await

    }

    async fn set_user_password<'user, U: Into<NamedReference<'user>> + Send + Sync>(

        &self,

        user: U,

        password: bonsaidb_core::connection::SensitiveString,

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

        self.data.storage.set_user_password(user, password).await

    }

    async fn authenticate<'user, U: Into<NamedReference<'user>> + Send + Sync>(

        &self,

        user: U,

        authentication: Authentication,

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

        self.data.storage.authenticate(user, authentication).await

    }

    async fn add_permission_group_to_user<

        'user,

        'group,

        U: Into<NamedReference<'user>> + Send + Sync,

        G: Into<NamedReference<'group>> + Send + Sync,

    >(

        &self,

        user: U,

        permission_group: G,

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

        self.data

            .storage

            .add_permission_group_to_user(user, permission_group)

            .await

    }

    async fn remove_permission_group_from_user<

        'user,

        'group,

        U: Into<NamedReference<'user>> + Send + Sync,

        G: Into<NamedReference<'group>> + Send + Sync,

    >(

        &self,

        user: U,

        permission_group: G,

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

        self.data

            .storage

            .remove_permission_group_from_user(user, permission_group)

            .await

    }

    async fn add_role_to_user<

        'user,

        'group,

        U: Into<NamedReference<'user>> + Send + Sync,

        G: Into<NamedReference<'group>> + Send + Sync,

    >(

        &self,

        user: U,

        role: G,

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

        self.data.storage.add_role_to_user(user, role).await

    }

    async fn remove_role_from_user<

        'user,

        'group,

        U: Into<NamedReference<'user>> + Send + Sync,

        G: Into<NamedReference<'group>> + Send + Sync,

    >(

        &self,

        user: U,

        role: G,

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

        self.data.storage.remove_role_from_user(user, role).await

    }

#[derive(Dispatcher, Debug)]

    async fn handle_subaction(

        &self,

        permissions: &Permissions,

        subaction: Self::Subaction,

    ) -> Result<Response<CustomApiResult<B::CustomApi>>, Error> {

        let dispatcher =

            <B::CustomApiDispatcher as CustomApiDispatcher<B>>::new(self.server, self.client);

        match dispatcher.dispatch(permissions, subaction).await {

            Ok(response) => Ok(Response::Api(Ok(response))),

            Err(err) => match err {

                BackendError::Server(server) => Err(server),

    }

    async fn handle(

        &self,

        permissions: &Permissions,

        request: ServerRequest,

    ) -> Result<Response<CustomApiResult<B::CustomApi>>, Error> {

        ServerRequestDispatcher::dispatch_to_handlers(self, permissions, request).await

    }

    async fn handle(

        &self,

        permissions: &Permissions,

        database_name: String,

        request: DatabaseRequest,

    ) -> Result<Response<CustomApiResult<B::CustomApi>>, Error> {

        let database = self

            .server

            .database_without_schema_internal(&database_name)

            .await?;

        DatabaseDispatcher {

            name: database_name,

            database,

            server_dispatcher: self,

        }

        .dispatch(permissions, request)

        .await

    }

    async fn resource_name<'a>(

        &'a self,

        database: &'a bonsaidb_core::connection::Database,

        _only_if_needed: &'a bool,

    ) -> Result<ResourceName<'a>, Error> {

        Ok(database_resource_name(&database.name))

    }

    fn action() -> Self::Action {

        BonsaiAction::Server(ServerAction::CreateDatabase)

    }

    async fn handle_protected(

        &self,

        _permissions: &Permissions,

        database: bonsaidb_core::connection::Database,

        only_if_needed: bool,

    ) -> Result<Response<CustomApiResult<B::CustomApi>>, Error> {

        self.server

            .create_database_with_schema(&database.name, database.schema, only_if_needed)

            .await?;

        Ok(Response::Server(ServerResponse::DatabaseCreated {

            name: database.name.clone(),

        }))

    }

    async fn resource_name<'a>(&'a self, database: &'a String) -> Result<ResourceName<'a>, Error> {

        Ok(database_resource_name(database))

    }

    fn action() -> Self::Action {

        BonsaiAction::Server(ServerAction::DeleteDatabase)

    }

    async fn handle_protected(

        &self,

        _permissions: &Permissions,

        name: String,

    ) -> Result<Response<CustomApiResult<B::CustomApi>>, Error> {

        self.server.delete_database(&name).await?;

        Ok(Response::Server(ServerResponse::DatabaseDeleted { name }))

    }

    async fn resource_name<'a>(&'a self) -> Result<ResourceName<'a>, Error> {

        Ok(bonsaidb_resource_name())

    }

    fn action() -> Self::Action {

        BonsaiAction::Server(ServerAction::ListDatabases)

    }

    async fn handle_protected(

        &self,

        _permissions: &Permissions,

    ) -> Result<Response<CustomApiResult<B::CustomApi>>, Error> {

            self.server.list_databases().await?,

    }

    async fn resource_name<'a>(&'a self) -> Result<ResourceName<'a>, Error> {

        Ok(bonsaidb_resource_name())

    }

    fn action() -> Self::Action {

        BonsaiAction::Server(ServerAction::ListAvailableSchemas)

    }

    async fn handle_protected(

        &self,

        _permissions: &Permissions,

    ) -> Result<Response<CustomApiResult<B::CustomApi>>, Error> {

            self.server.list_available_schemas().await?,

    }

    async fn resource_name<'a>(&'a self, _username: &'a String) -> Result<ResourceName<'a>, Error> {

        Ok(bonsaidb_resource_name())

    }

    fn action() -> Self::Action {

        BonsaiAction::Server(ServerAction::CreateUser)

    }

    async fn handle_protected(

        &self,

        _permissions: &Permissions,

        username: String,

    ) -> Result<Response<CustomApiResult<B::CustomApi>>, Error> {

            id: self.server.create_user(&username).await?,

    }

    async fn verify_permissions(

        &self,

        permissions: &Permissions,

        user: &NamedReference<'static>,

        authentication: &Authentication,

    ) -> Result<(), Error> {

        let id = user

            .id::<User, _>(&self.server.admin().await)

            .await?

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

        match authentication {

            Authentication::Password(_) => {

                permissions.check(

                    user_resource_name(id),

                    &BonsaiAction::Server(ServerAction::Authenticate(

                        AuthenticationMethod::PasswordHash,

                    )),

                )?;

                Ok(())

    }

    async fn handle_protected(

        &self,

        _permissions: &Permissions,

        username: NamedReference<'static>,

        authentication: Authentication,

    ) -> Result<Response<CustomApiResult<B::CustomApi>>, Error> {

        let mut response = self

            .server

            .authenticate(username.clone(), authentication)

            .await?;

        response.permissions = Permissions::merged([

            &response.permissions,

            &self.server.data.authenticated_permissions,

            &self.server.data.default_permissions,

        ]);

        self.client

            .logged_in_as(response.user_id, response.permissions.clone())

        Ok(Response::Server(ServerResponse::Authenticated(response)))

    }

    async fn resource_name<'a>(

        &'a self,

        user: &'a NamedReference<'static>,

        _group: &'a NamedReference<'static>,

        _should_be_member: &'a bool,

    ) -> Result<ResourceName<'a>, Error> {

        let id = user

            .id::<User, _>(&self.server.admin().await)

            .await?

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

        Ok(user_resource_name(id))

    }