#[derive_where(Clone)]

#[derive(Default)]

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

        &self.0

    }

    pub async fn open(

        configuration: ServerConfiguration<B>,

    ) -> Result<Self, BackendError<B::Error>> {

        let configuration = register_api_handlers(B::configure(configuration)?)?;

        let (request_sender, request_receiver) = flume::unbounded::<ClientRequest<B>>();

        for _ in 0..configuration.request_workers {

            let request_receiver = request_receiver.clone();

            tokio::task::spawn(async move {

                while let Ok(mut client_request) = request_receiver.recv_async().await {

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

                    let session = client_request.session.clone();

                    let result = match client_request.server.storage.assume_session(session) {

                        Ok(storage) => {

                            let client = HandlerSession {

                                server: &client_request.server,

                                client: &client_request.client,

                                as_client: Self {

                                    data: client_request.server.data.clone(),

                                    storage,

                                },

                            };

                            ServerDispatcher::dispatch_api_request(

                                client,

                                &request.name,

                                request.value.unwrap(),

                            )

                            .await

                            .map_err(bonsaidb_core::Error::from)

                        Err(err) => Err(err),

                    drop(client_request.result_sender.send((request.name, result)));

            });

        }

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

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

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

        let server = Self {

            storage,

            data: Arc::new(Data {

                backend: configuration.backend,

                clients: parking_lot::RwLock::default(),

                request_processor: request_sender,

                default_session: Session {

                    permissions: default_permissions,

                    ..Session::default()

                },

                client_simultaneous_request_limit: configuration.client_simultaneous_request_limit,

                primary_tls_key: CachedCertifiedKey::default(),

                primary_domain: configuration.server_name,

                custom_apis: parking_lot::RwLock::new(configuration.custom_apis),

                #[cfg(feature = "acme")]

                acme: configuration.acme,

                #[cfg(feature = "acme")]

                alpn_keys: AlpnKeys::default(),

                shutdown: Shutdown::new(),

            }),

        };

        server.data.backend.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 fn backend(&self) -> &B {

        &self.data.backend

    }

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

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

        ServerDatabase {

            server: self.clone(),

            db,

        }

    }

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

        let db = self.storage.database::<Hosted>("_hosted").await.unwrap();

        ServerDatabase {

            server: self.clone(),

            db,

        }

    }

    pub(crate) fn custom_api_dispatcher(&self, name: &ApiName) -> Option<Arc<dyn AnyHandler<B>>> {

        let dispatchers = self.data.custom_apis.read();

        dispatchers.get(name).cloned()

    }

    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::other("bonsaidb-server config", "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_async(&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)

            .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)

            .query()

            .await?

            .into_iter()

            .next()

            Ok(mapping.value)

            Err(Error::Core(bonsaidb_core::Error::other(

                "bonsaidb-server config",

                format!("no certificate found for {}", self.data.primary_domain),

            )))

    }

    pub async fn listen_on(&self, config: impl Into<BonsaiListenConfig>) -> Result<(), Error> {

        let config = config.into();

        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(config.address);

        builder.set_max_idle_timeout(None)?;

        builder.set_server_key_pair(Some(keypair));

        builder.set_reuse_address(config.reuse_address);

        let mut server = builder.build()?;

        let mut shutdown_watcher = self

            .data

            .shutdown

            .watcher()

            .expect("server already shut down");

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

            shutdown_state = shutdown_watcher.wait_for_shutdown() => {

            msg = server.next() => msg

            let address = incoming.remote_address();

            let connection = match incoming.accept::<()>().await {

                Ok(connection) => connection,

            let task_self = self.clone();

            tokio::spawn(async move {

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

                }

            });

        Ok(())

    }

    pub fn connected_clients(&self) -> Vec<ConnectedClient<B>> {

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

        clients.values().cloned().collect()

    }

    async fn initialize_client(

        &self,

        transport: Transport,

        address: SocketAddr,

        sender: Sender<(Option<SessionId>, ApiName, Bytes)>,

    ) -> Option<OwnedClient<B>> {

        if !self.data.default_session.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 = self.data.clients.write();

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

                let client = OwnedClient::new(

                    next_id,

                    address,

                    transport,

                    sender,

                    self.clone(),

                    self.data.default_session.clone(),

                );

                e.insert(client.clone());

                break client;

        match self.data.backend.client_connected(&client, self).await {

            Ok(ConnectionHandling::Accept) => Some(client),

    }

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

        let removed_client = {

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

            clients.remove(&id)

        if let Some(client) = removed_client {

            client.set_disconnected();

            for session in client.all_sessions::<Vec<_>>() {

                if let Err(err) = self

                    .data

                    .backend

                    .client_session_ended(session, &client, true, self)

                }

            if let Err(err) = self.data.backend.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, networking::Payload>()

                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((session_id, name, bytes)) =

                                api_response_receiver.recv_async().await

                                if task_sender

                                    .send(&Payload {

                                        id: None,

                                        session_id,

                                        name,

                                        value: Ok(bytes),

                                    })

                                    .is_err()

                                }

                        });

                        let task_self = self.clone();

                        let Some(shutdown) = self.data.shutdown.watcher().await else {

                        tokio::spawn(async move {

                            if let Err(err) = task_self

                                .handle_stream(disconnector, sender, receiver, shutdown)

                                .await

                            }

                        });

        Ok(())

    }

    async fn handle_client_requests(

        &self,

        client: ConnectedClient<B>,

        request_receiver: flume::Receiver<Payload>,

        response_sender: flume::Sender<Payload>,

        mut shutdown: ShutdownStateWatcher,

    ) {

        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 = 'payload: loop {

                    tokio::select! {

                        payload = request_receiver.recv_async() => {

                        state = shutdown.wait_for_shutdown() => {

                    }

                };

                let session_id = payload.session_id;

                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,

                    move |name, value| async move {

                        drop(task_sender.send(Payload {

                            session_id,

                            id,

                            name,

                            value,

                        }));

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

                        notify.notify_one();

                        Ok(())

                    },

                    client.clone(),

                )

                .unwrap();

            }

    }

    fn handle_request_through_worker<

        F: FnOnce(ApiName, Result<Bytes, bonsaidb_core::Error>) -> R + Send + 'static,

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

    >(

        &self,

        request: Payload,

        callback: F,

        client: ConnectedClient<B>,

    ) -> Result<(), Error> {

        let (result_sender, result_receiver) = oneshot::channel();

        let session = client

            .session(request.session_id)

            .unwrap_or_else(|| self.data.default_session.clone());

        self.data

            .request_processor

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

                request,

                self.clone(),

                client,

                session,

                result_sender,

            ))

            .map_err(|_| Error::InternalCommunication)?;

        tokio::spawn(async move {

            let (name, result) = result_receiver.await?;

            callback(name, result).await?;

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

        });

        Ok(())

    }

    async fn handle_stream(

        &self,

        client: OwnedClient<B>,

        sender: fabruic::Sender<Payload>,

        mut receiver: fabruic::Receiver<Payload>,

        mut shutdown: ShutdownStateWatcher,

    ) -> Result<(), Error> {

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

        tokio::spawn({

            let mut shutdown = shutdown.clone();

            async move {

                    let payload = loop {

                        tokio::select! {

                            payload = payload_receiver.recv_async() => {

                            shutdown = shutdown.wait_for_shutdown() => {

                        }

                    };

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

                    }

            }

        });

        let (request_sender, request_receiver) =

            flume::bounded::<Payload>(self.data.client_simultaneous_request_limit);

        let task_self = self.clone();

        tokio::spawn({

            let shutdown = shutdown.clone();

            async move {

                task_self

                    .handle_client_requests(

                        client.clone(),

                        request_receiver,

                        payload_sender,

                        shutdown,

                    )

                    .await;

            }

        });

            let payload = loop {

                tokio::select! {

                    payload = receiver.next() => {

                    shutdown = shutdown.wait_for_shutdown() => {

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

    }

    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(async_lock::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 {

                } else if self.data.shutdown.should_shutdown() {

                }

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

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

        &self.storage

    }

    pub fn new(

        request: Payload,

        server: CustomServer<B>,

        client: ConnectedClient<B>,

        session: Session,

        result_sender: oneshot::Sender<(ApiName, Result<Bytes, bonsaidb_core::Error>)>,

    ) -> Self {

        Self {

            request: Some(request),

            server,

            client,

            session,

            result_sender,

        }

    }

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

        self.storage.session()

    }

    async fn create_database_with_schema(

        &self,

        name: &str,

        schema: SchemaName,

        only_if_needed: bool,

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

        self.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.storage.database::<DB>(name).await?;

        Ok(ServerDatabase {

            server: self.clone(),

            db,

        })

    }

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

        self.storage.delete_database(name).await

    }

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

        self.storage.list_databases().await

    }

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

        self.storage.list_available_schemas().await

    }

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

        self.storage.create_user(username).await

    }

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

        &self,

        user: U,

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

        self.storage.delete_user(user).await

    }

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

        &self,

        user: U,

        password: bonsaidb_core::connection::SensitiveString,

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

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

    }

    async fn authenticate(

        &self,

        authentication: bonsaidb_core::connection::Authentication,

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

        let storage = self.storage.authenticate(authentication).await?;

        Ok(Self {

            data: self.data.clone(),

            storage,

        })

    }

    async fn assume_identity(

        &self,

        identity: IdentityReference<'_>,

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

        let storage = self.storage.assume_identity(identity).await?;

        Ok(Self {

            data: self.data.clone(),

            storage,

        })

    }

    async 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.storage

            .add_permission_group_to_user(user, permission_group)

            .await

    }

    async 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.storage

            .remove_permission_group_from_user(user, permission_group)

            .await

    }

    async 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.storage.add_role_to_user(user, role).await

    }

    async 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.storage.remove_role_from_user(user, role).await

    }

#[derive(Default)]