fn clone(&self) -> Self {

        Self {

            data: self.data.clone(),

        }

    }

    pub(crate) async fn new<DB: Schema, S: Into<Cow<'static, str>> + Send>(

        name: S,

        context: Context,

        storage: Storage,

    ) -> Result<Self, Error> {

        let name = name.into();

        let schema = Arc::new(DB::schematic()?);

        let db = Self {

            data: Arc::new(Data {

                name: Arc::new(name),

                context,

                storage: storage.clone(),

                schema,

                effective_permissions: None,

            }),

        };

        if db.data.storage.check_view_integrity_on_database_open() {

            for view in db.data.schema.views() {

                db.data

                    .storage

                    .tasks()

                    .spawn_integrity_check(view, &db)

        }

        storage.tasks().spawn_key_value_expiration_loader(&db).await;

        Ok(db)

    }

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

        self.data.name.as_ref()

    }

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

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

        storage.create_database::<DB>("default", true).await?;

        Ok(storage.database::<DB>("default").await?)

    }

    pub fn storage(&self) -> &'_ Storage {

        &self.data.storage

    }

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

        &self.data.schema

    }

    pub(crate) fn roots(&self) -> &'_ nebari::Roots<AnyFile> {

        &self.data.context.roots

    }

    async fn for_each_in_view<

        F: FnMut(ViewEntryCollection) -> Result<(), bonsaidb_core::Error> + Send + Sync,

    >(

        &self,

        view: &dyn view::Serialized,

        key: Option<QueryKey<Bytes>>,

        order: Sort,

        limit: Option<usize>,

        access_policy: AccessPolicy,

        mut callback: F,

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

        if matches!(access_policy, AccessPolicy::UpdateBefore) {

            self.data

                .storage

                .tasks()

                .update_view_if_needed(view, self)

                .await?;

        }

        let view_entries = self

            .roots()

            .tree(self.collection_tree(

                &view.collection(),

                view_entries_tree_name(&view.view_name()),

            )?)

            .map_err(Error::from)?;

            for entry in Self::create_view_iterator(&view_entries, key, order, limit)? {

                callback(entry)?;

        if matches!(access_policy, AccessPolicy::UpdateAfter) {

            let db = self.clone();

            let view_name = view.view_name();

            tokio::task::spawn(async move {

                let view = db

                    .data

                    .schema

                    .view_by_name(&view_name)

                    .expect("query made with view that isn't registered with this database");

                db.data

                    .storage

                    .tasks()

                    .update_view_if_needed(view, &db)

                    .await

            });

        }

        Ok(())

    }

    async fn for_each_view_entry<

        V: schema::View,

        F: FnMut(ViewEntryCollection) -> Result<(), bonsaidb_core::Error> + Send + Sync,

    >(

        &self,

        key: Option<QueryKey<V::Key>>,

        order: Sort,

        limit: Option<usize>,

        access_policy: AccessPolicy,

        callback: F,

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

        let view = self

            .data

            .schema

            .view::<V>()

            .expect("query made with view that isn't registered with this database");

        self.for_each_in_view(

            view,

            key.map(|key| key.serialized()).transpose()?,

            order,

            limit,

            access_policy,

            callback,

        )

        .await

    }

    pub async fn internal_get_from_collection_id(

        &self,

        id: DocumentId,

        collection: &CollectionName,

    ) -> Result<Option<OwnedDocument>, bonsaidb_core::Error> {

        self.get_from_collection_id(id, collection).await

    }

    pub async fn list_from_collection(

        &self,

        ids: Range<DocumentId>,

        order: Sort,

        limit: Option<usize>,

        collection: &CollectionName,

    ) -> Result<Vec<OwnedDocument>, bonsaidb_core::Error> {

        self.list(ids, order, limit, collection).await

    }

    pub async fn internal_get_multiple_from_collection_id(

        &self,

        ids: &[DocumentId],

        collection: &CollectionName,

    ) -> Result<Vec<OwnedDocument>, bonsaidb_core::Error> {

        self.get_multiple_from_collection_id(ids, collection).await

    }

    pub async fn compact_collection_by_name(

        &self,

        collection: CollectionName,

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

        self.storage()

            .tasks()

            .compact_collection(self.clone(), collection)

            .await?;

        Ok(())

    }

    pub async fn query_by_name(

        &self,

        view: &ViewName,

        key: Option<QueryKey<Bytes>>,

        order: Sort,

        limit: Option<usize>,

        access_policy: AccessPolicy,

    ) -> Result<Vec<bonsaidb_core::schema::view::map::Serialized>, bonsaidb_core::Error> {

        if let Some(view) = self.schematic().view_by_name(view) {

            let mut results = Vec::new();

            self.for_each_in_view(view, key, order, limit, access_policy, |collection| {

                let entry = ViewEntry::from(collection);

                for mapping in entry.mappings {

                    results.push(bonsaidb_core::schema::view::map::Serialized {

                        source: mapping.source,

                        key: entry.key.clone(),

                        value: mapping.value,

                    });

                }

                Ok(())

            })

            .await?;

            Ok(results)

    }

    pub async fn query_by_name_with_docs(

        &self,

        view: &ViewName,

        key: Option<QueryKey<Bytes>>,

        order: Sort,

        limit: Option<usize>,

        access_policy: AccessPolicy,

    ) -> Result<bonsaidb_core::schema::view::map::MappedSerializedDocuments, bonsaidb_core::Error>

    {

        let results = self

            .query_by_name(view, key, order, limit, access_policy)

            .await?;

        let view = self.schematic().view_by_name(view).unwrap(); // query() will fail if it's not present

        let documents = self

            .get_multiple_from_collection_id(

                &results.iter().map(|m| m.source.id).collect::<Vec<_>>(),

                &view.collection(),

            )

            .await?

            .into_iter()

            .map(|doc| (doc.header.id, doc))

            .collect::<BTreeMap<_, _>>();

        Ok(

            bonsaidb_core::schema::view::map::MappedSerializedDocuments {

                mappings: results,

                documents,

            },

        )

    }

    pub async fn reduce_by_name(

        &self,

        view: &ViewName,

        key: Option<QueryKey<Bytes>>,

        access_policy: AccessPolicy,

    ) -> Result<Vec<u8>, bonsaidb_core::Error> {

        self.reduce_in_view(view, key, access_policy).await

    }

    pub async fn reduce_grouped_by_name(

        &self,

        view: &ViewName,

        key: Option<QueryKey<Bytes>>,

        access_policy: AccessPolicy,

    ) -> Result<Vec<MappedSerializedValue>, bonsaidb_core::Error> {

        self.grouped_reduce_in_view(view, key, access_policy).await

    }

    pub async fn delete_docs_by_name(

        &self,

        view: &ViewName,

        key: Option<QueryKey<Bytes>>,

        access_policy: AccessPolicy,

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

        let view = self

            .data

            .schema

            .view_by_name(view)

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

        let collection = view.collection();

        let mut transaction = Transaction::default();

        self.for_each_in_view(view, key, Sort::Ascending, None, access_policy, |entry| {

            let entry = ViewEntry::from(entry);

            for mapping in entry.mappings {

                transaction.push(Operation::delete(collection.clone(), mapping.source));

            }

            Ok(())

        })

        .await?;

        let results = Connection::apply_transaction(self, transaction).await?;

        Ok(results.len() as u64)

    }

    async fn get_from_collection_id(

        &self,

        id: DocumentId,

        collection: &CollectionName,

    ) -> Result<Option<OwnedDocument>, bonsaidb_core::Error> {

        let task_self = self.clone();

        let collection = collection.clone();

        tokio::task::spawn_blocking(move || {

            let tree = task_self

                .data

                .context

                .roots

                .tree(task_self.collection_tree::<Versioned, _>(

                    &collection,

                    document_tree_name(&collection),

                )?)

                .map_err(Error::from)?;

            if let Some(vec) = tree.get(id.as_ref()).map_err(Error::from)? {

                Ok(Some(deserialize_document(&vec)?.into_owned()))

                Ok(None)

        })

        .await

        .unwrap()

    }

    async fn get_multiple_from_collection_id(

        &self,

        ids: &[DocumentId],

        collection: &CollectionName,

    ) -> Result<Vec<OwnedDocument>, bonsaidb_core::Error> {

        let task_self = self.clone();

        let ids = ids.to_vec();

        let collection = collection.clone();

        tokio::task::spawn_blocking(move || {

            let tree = task_self

                .data

                .context

                .roots

                .tree(task_self.collection_tree::<Versioned, _>(

                    &collection,

                    document_tree_name(&collection),

                )?)

                .map_err(Error::from)?;

            let mut ids = ids.iter().map(DocumentId::deref).collect::<Vec<_>>();

            ids.sort();

            let keys_and_values = tree.get_multiple(&ids).map_err(Error::from)?;

            keys_and_values

                .into_iter()

                .map(|(_, value)| deserialize_document(&value).map(BorrowedDocument::into_owned))

                .collect::<Result<Vec<_>, Error>>()

        })

        .await

        .unwrap()

        .map_err(bonsaidb_core::Error::from)

    }

    pub(crate) async fn list(

        &self,

        ids: Range<DocumentId>,

        sort: Sort,

        limit: Option<usize>,

        collection: &CollectionName,

    ) -> Result<Vec<OwnedDocument>, bonsaidb_core::Error> {

        let task_self = self.clone();

        let collection = collection.clone();

        tokio::task::spawn_blocking(move || {

            let tree = task_self

                .data

                .context

                .roots

                .tree(task_self.collection_tree::<Versioned, _>(

                    &collection,

                    document_tree_name(&collection),

                )?)

                .map_err(Error::from)?;

            let mut found_docs = Vec::new();

            let mut keys_read = 0;

            let ids = DocumentIdRange(ids);

            tree.scan(

                &ids.borrow_as_bytes(),

                match sort {

                    Sort::Ascending => true,

                    Sort::Descending => false,

                |_, _| {

                    if let Some(limit) = limit {

                        if keys_read >= limit {

                            return KeyEvaluation::Stop;

                        }

                        keys_read += 1;

                    }

                    KeyEvaluation::ReadData

                },

                |_, _, doc| {

                        deserialize_document(&doc)

                            .map(BorrowedDocument::into_owned)

                            .map_err(AbortError::Other)?,

                    Ok(())

                },

            .map_err(|err| match err {

            })

            .unwrap();

            Ok(found_docs)

        })

        .await

        .unwrap()

    }

    async fn reduce_in_view(

        &self,

        view_name: &ViewName,

        key: Option<QueryKey<Bytes>>,

        access_policy: AccessPolicy,

    ) -> Result<Vec<u8>, bonsaidb_core::Error> {

        let view = self

            .data

            .schema

            .view_by_name(view_name)

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

        let mut mappings = self

            .grouped_reduce_in_view(view_name, key, access_policy)

            .await?;

        let result = if mappings.len() == 1 {

            mappings.pop().unwrap().value.into_vec()

            view.reduce(

                &mappings

                    .iter()

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

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

                true,

            )

            .map_err(Error::from)?

        Ok(result)

    }

    async fn grouped_reduce_in_view(

        &self,

        view_name: &ViewName,

        key: Option<QueryKey<Bytes>>,

        access_policy: AccessPolicy,

    ) -> Result<Vec<MappedSerializedValue>, bonsaidb_core::Error> {

        let view = self

            .data

            .schema

            .view_by_name(view_name)

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

        let mut mappings = Vec::new();

        self.for_each_in_view(view, key, Sort::Ascending, None, access_policy, |entry| {

            let entry = ViewEntry::from(entry);

            mappings.push(MappedSerializedValue {

                key: entry.key,

                value: entry.reduced_value,

            });

            Ok(())

        })

        .await?;

        Ok(mappings)

    }

    fn apply_transaction_to_roots(

        &self,

        transaction: &Transaction,

    ) -> Result<Vec<OperationResult>, Error> {

        let mut open_trees = OpenTrees::default();

        for op in &transaction.operations {

            if !self.data.schema.contains_collection_id(&op.collection) {

                return Err(Error::Core(bonsaidb_core::Error::CollectionNotFound));

            }

            let vault = if let Some(encryption_key) =

                self.collection_encryption_key(&op.collection).cloned()

                if let Some(mut vault) = self.storage().tree_vault().cloned() {

                    vault.key = Some(encryption_key);

                    Some(vault)

                    TreeVault::new_if_needed(

                        Some(encryption_key),

                        self.storage().vault(),

                        #[cfg(feature = "compression")]

                        None,

                    )

                self.storage().tree_vault().cloned()

            open_trees.open_trees_for_document_change(

                &op.collection,

                &self.data.schema,

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

                vault,

            );

        let mut roots_transaction = self

            .data

            .context

            .roots

            .transaction::<_, dyn AnyTreeRoot<AnyFile>>(&open_trees.trees)?;

        let mut results = Vec::new();

        let mut changed_documents = Vec::new();

        for op in &transaction.operations {

            let result = self.execute_operation(

                op,

                &mut roots_transaction,

                &open_trees.trees_index_by_name,

            )?;

            match &result {

                OperationResult::DocumentUpdated { header, collection } => {

                    changed_documents.push(ChangedDocument {

                        collection: collection.clone(),

                        id: header.id,

                        deleted: false,

                    });

                }

                OperationResult::DocumentDeleted { id, collection } => {

                    changed_documents.push(ChangedDocument {

                        collection: collection.clone(),

                        id: *id,

                        deleted: true,

                    });

                }

            results.push(result);

        for (collection, changed_documents) in &changed_documents

            .iter()

            .group_by(|doc| doc.collection.clone())

            if let Some(views) = self.data.schema.views_in_collection(&collection) {

                let changed_documents = changed_documents.collect::<Vec<_>>();

                for view in views {

                    if !view.unique() {

                        let view_name = view.view_name();

                        for changed_document in &changed_documents {

                            let invalidated_docs = roots_transaction

                                .tree::<Unversioned>(

                                    open_trees.trees_index_by_name

                                        [&view_invalidated_docs_tree_name(&view_name)],

                                )

                                .unwrap();

                            invalidated_docs.set(changed_document.id.as_ref().to_vec(), b"")?;

                    }

            }

        roots_transaction

            .entry_mut()

            .set_data(pot::to_vec(&Changes::Documents(changed_documents))?)?;

        roots_transaction.commit()?;

        Ok(results)

    }

    fn execute_operation(

        &self,

        operation: &Operation,

        transaction: &mut ExecutingTransaction<AnyFile>,

        tree_index_map: &HashMap<String, usize>,

    ) -> Result<OperationResult, Error> {

        match &operation.command {

            Command::Insert { id, contents } => {

                self.execute_insert(operation, transaction, tree_index_map, *id, contents)

            Command::Update { header, contents } => self.execute_update(

                operation,

                transaction,

                tree_index_map,

                header.id,

                Some(&header.revision),

                contents,

            ),

            Command::Overwrite { id, contents } => {

                self.execute_update(operation, transaction, tree_index_map, *id, None, contents)

            Command::Delete { header } => {

                self.execute_delete(operation, transaction, tree_index_map, header)

    }

    fn execute_update(

        &self,

        operation: &Operation,

        transaction: &mut ExecutingTransaction<AnyFile>,

        tree_index_map: &HashMap<String, usize>,

        id: DocumentId,

        check_revision: Option<&Revision>,

        contents: &[u8],

    ) -> Result<OperationResult, crate::Error> {

        let documents = transaction

            .tree::<Versioned>(tree_index_map[&document_tree_name(&operation.collection)])

            .unwrap();

        let document_id = ArcBytes::from(id.as_ref());

        let mut result = None;

        let mut updated = false;

        documents.modify(

            vec![document_id.clone()],

            nebari::tree::Operation::CompareSwap(CompareSwap::new(&mut |_key,

                if let Some(old) = value {

                    let doc = match deserialize_document(&old) {

                        Ok(doc) => doc,

                    if check_revision.is_none() || Some(&doc.header.revision) == check_revision {

                        if let Some(updated_revision) = doc.header.revision.next_revision(contents)

                            let updated_header = Header {

                                id,

                                revision: updated_revision,

                            };

                            let serialized_doc = match serialize_document(&BorrowedDocument {

                                header: updated_header.clone(),

                                contents: CowBytes::from(contents),

                            }) {

                                Ok(bytes) => bytes,

                            result = Some(Ok(OperationResult::DocumentUpdated {

                                collection: operation.collection.clone(),

                                header: updated_header,

                            }));

                            updated = true;

                            return nebari::tree::KeyOperation::Set(ArcBytes::from(serialized_doc));

                        }

                        // If no new revision was made, it means an attempt to

                        // save a document with the same contents was made.

                        // We'll return a success but not actually give a new

                        // version

                        result = Some(Ok(OperationResult::DocumentUpdated {

                            collection: operation.collection.clone(),

                            header: doc.header,

                        }));

                    } else {

                        result = Some(Err(Error::Core(bonsaidb_core::Error::DocumentConflict(

                            operation.collection.clone(),

                            Box::new(doc.header),

                        ))));

                    }

                } else if check_revision.is_none() {

                } else {

                    result = Some(Err(Error::Core(bonsaidb_core::Error::DocumentNotFound(

                        operation.collection.clone(),

                        Box::new(id),

                    ))));

                }

                nebari::tree::KeyOperation::Skip

            })),

        )?;

        if updated {

            self.update_unique_views(&document_id, operation, transaction, tree_index_map)?;

        }

        result.expect("nebari should invoke the callback even when the key isn't found")

    }

    fn execute_insert(

        &self,

        operation: &Operation,

        transaction: &mut ExecutingTransaction<AnyFile>,

        tree_index_map: &HashMap<String, usize>,

        id: Option<DocumentId>,

        contents: &[u8],

    ) -> Result<OperationResult, Error> {

        let documents = transaction

            .tree::<Versioned>(tree_index_map[&document_tree_name(&operation.collection)])

            .unwrap();

        let id = if let Some(id) = id {

            id

        } else if let Some(last_key) = documents.last_key()? {

            let id = DocumentId::try_from(last_key.as_slice())?;

            self.data

                .schema

                .next_id_for_collection(&operation.collection, Some(id))?

            self.data

                .schema

                .next_id_for_collection(&operation.collection, None)?

        let doc = BorrowedDocument::new(id, contents);

        let serialized: Vec<u8> = serialize_document(&doc)?;

        let document_id = ArcBytes::from(doc.header.id.as_ref().to_vec());

        if let Some(document) = documents.replace(document_id.clone(), serialized)? {

            let doc = deserialize_document(&document)?;

            Err(Error::Core(bonsaidb_core::Error::DocumentConflict(

                operation.collection.clone(),

                Box::new(doc.header),

            )))

            self.update_unique_views(&document_id, operation, transaction, tree_index_map)?;

            Ok(OperationResult::DocumentUpdated {

                collection: operation.collection.clone(),

                header: doc.header,

            })

    }

    fn execute_delete(

        &self,

        operation: &Operation,

        transaction: &mut ExecutingTransaction<AnyFile>,

        tree_index_map: &HashMap<String, usize>,

        header: &Header,

    ) -> Result<OperationResult, Error> {

        let documents = transaction

            .tree::<Versioned>(tree_index_map[&document_tree_name(&operation.collection)])

            .unwrap();

        if let Some(vec) = documents.remove(header.id.as_ref())? {

            let doc = deserialize_document(&vec)?;

            if &doc.header == header {

                self.update_unique_views(

                    doc.header.id.as_ref(),

                    operation,

                    transaction,

                    tree_index_map,

                )?;

                Ok(OperationResult::DocumentDeleted {

                    collection: operation.collection.clone(),

                    id: header.id,

                })

    }

        if let Some(unique_views) = self

            .data

            .schema

            .unique_views_in_collection(&operation.collection)

            for view in unique_views {

                let name = view.view_name();

                mapper::DocumentRequest {

                    database: self,

                    document_id,

                    map_request: &mapper::Map {

                        database: self.data.name.clone(),

                        collection: operation.collection.clone(),

                        view_name: name.clone(),

                    },

                    transaction,

                    document_map_index: tree_index_map[&view_document_map_tree_name(&name)],

                    documents_index: tree_index_map[&document_tree_name(&operation.collection)],

                    omitted_entries_index: tree_index_map[&view_omitted_docs_tree_name(&name)],

                    view_entries_index: tree_index_map[&view_entries_tree_name(&name)],

                    view,

                }

                .map()?;

        }

        Ok(())

    }

    fn create_view_iterator<'a, K: for<'k> Key<'k> + 'a>(

        view_entries: &'a Tree<Unversioned, AnyFile>,

        key: Option<QueryKey<K>>,

        order: Sort,

        limit: Option<usize>,

    ) -> Result<Vec<ViewEntryCollection>, Error> {

        let mut values = Vec::new();

        let forwards = match order {

            Sort::Ascending => true,

            Sort::Descending => false,

        let mut values_read = 0;

        if let Some(key) = key {

            match key {

                QueryKey::Range(range) => {

                    let range = range

                        .as_ord_bytes()

                        .map_err(view::Error::key_serialization)?;

                    view_entries.scan::<Infallible, _, _, _, _>(

                        &range.map_ref(|bytes| &bytes[..]),

                        forwards,

                        |_, _, _| true,

                        |_, _| {

                            if let Some(limit) = limit {

                                if values_read >= limit {

                                    return KeyEvaluation::Stop;

                                }

                                values_read += 1;

                            }

                            KeyEvaluation::ReadData

                        },

                        |_key, _index, value| {

                            values.push(value);

                            Ok(())

                        },

                    )?;

                QueryKey::Matches(key) => {

                    let key = key

                        .as_ord_bytes()

                        .map_err(view::Error::key_serialization)?

                        .to_vec();

                    values.extend(view_entries.get(&key)?);

                QueryKey::Multiple(list) => {

                    let mut list = list

                        .into_iter()

                        .map(|key| {

                            key.as_ord_bytes()

                                .map(|bytes| bytes.to_vec())

                                .map_err(view::Error::key_serialization)

                        })

                        .collect::<Result<Vec<_>, _>>()?;

                    list.sort();

                    values.extend(

                        view_entries

                            .get_multiple(&list.iter().map(Vec::as_slice).collect::<Vec<_>>())?

                            .into_iter()

                            .map(|(_, value)| value),

            view_entries.scan::<Infallible, _, _, _, _>(

                &(..),

                forwards,

                |_, _, _| true,

                |_, _| {

                    if let Some(limit) = limit {

                    }

                    KeyEvaluation::ReadData

                },

                |_, _, value| {

                    values.push(value);

                    Ok(())

                },

            )?;

        values

            .into_iter()

            .map(|value| bincode::deserialize(&value).map_err(Error::from))

            .collect::<Result<Vec<_>, Error>>()

    }

    pub(crate) fn collection_encryption_key(&self, collection: &CollectionName) -> Option<&KeyId> {

        self.schematic()

            .encryption_key_for_collection(collection)

            .or_else(|| self.storage().default_encryption_key())

    }

    pub(crate) fn collection_tree<R: Root, S: Into<Cow<'static, str>>>(

        &self,

        collection: &CollectionName,

        name: S,

    ) -> Result<TreeRoot<R, AnyFile>, Error> {

        let mut tree = R::tree(name);

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

        match (

            self.collection_encryption_key(collection),

            self.storage().tree_vault().cloned(),

            (Some(override_key), Some(mut vault)) => {

                #[cfg(feature = "encryption")]

                {

                    vault.key = Some(override_key.clone());

                    tree = tree.with_vault(vault);

                }

            (None, Some(vault)) => {

                tree = tree.with_vault(vault);

            }

            (key, None) => {

                if let Some(vault) = TreeVault::new_if_needed(

                    key.cloned(),

                    self.storage().vault(),

                    #[cfg(feature = "compression")]

                    None,

                ) {

                    tree = tree.with_vault(vault);

                }

        Ok(tree)

    }

    pub(crate) async fn update_key_expiration_async<'key>(

        &self,

        tree_key: impl Into<Cow<'key, str>>,

        expiration: Option<Timestamp>,

    ) {

        self.data

            .context

            .update_key_expiration_async(tree_key, expiration)

    }

pub(crate) fn deserialize_document(bytes: &[u8]) -> Result<BorrowedDocument<'_>, Error> {

    match pot::from_slice::<BorrowedDocument<'_>>(bytes) {

        Ok(document) => Ok(document),

}

fn serialize_document(document: &BorrowedDocument<'_>) -> Result<Vec<u8>, bonsaidb_core::Error> {

    pot::to_vec(document)

        .map_err(Error::from)

        .map_err(bonsaidb_core::Error::from)

}

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

    ) -> Result<Vec<OperationResult>, bonsaidb_core::Error> {

        let task_self = self.clone();

        let mut unique_view_tasks = Vec::new();

        for collection_name in transaction

            .operations

            .iter()

            .map(|op| &op.collection)

            .collect::<HashSet<_>>()

            if let Some(views) = self.data.schema.views_in_collection(collection_name) {

                for view in views {

                    if view.unique() {

                        if let Some(task) = self

                            .data

                            .storage

                            .tasks()

                            .spawn_integrity_check(view, self)

                        {

                            unique_view_tasks.push(task);

                        }

                    }

            }

        let mut unique_view_mapping_tasks = Vec::new();

        for task in unique_view_tasks {

            if let Some(spawned_task) = task

                .receive()

                .await

                .map_err(Error::from)?

                .map_err(Error::from)?

            {

                unique_view_mapping_tasks.push(spawned_task);

            }

        for task in unique_view_mapping_tasks {

            let mut task = fast_async_lock!(task);

            if let Some(task) = task.take() {

                task.receive()

                    .await

                    .map_err(Error::from)?

                    .map_err(Error::from)?;

        tokio::task::spawn_blocking(move || task_self.apply_transaction_to_roots(&transaction))

            .await

            .map_err(|err| bonsaidb_core::Error::Database(err.to_string()))?

            .map_err(bonsaidb_core::Error::from)

    }

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

    {

        self.get_from_collection_id(id.into().to_document_id()?, &C::collection_name())

            .await

    }

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

    {

            &ids.into_iter()

                .map(|id| id.into().to_document_id())

                .collect::<Result<Vec<_>, _>>()?,

            &C::collection_name(),

        )

        .await

    }

    #[cfg_attr(feature = "tracing", tracing::instrument(skip(ids, order, limit)))]

    {

            ids.into().map_result(|id| id.into().to_document_id())?,

            order,

            limit,

            &C::collection_name(),

        )

        .await

    }

        tracing::instrument(skip(starting_id, result_limit))

    ) -> Result<Vec<transaction::Executed>, bonsaidb_core::Error> {

        let result_limit = result_limit

            .unwrap_or(LIST_TRANSACTIONS_DEFAULT_RESULT_COUNT)

            .min(LIST_TRANSACTIONS_MAX_RESULTS);

        if result_limit > 0 {

            let task_self = self.clone();

            tokio::task::spawn_blocking::<_, Result<Vec<transaction::Executed>, Error>>(move || {

                let range = if let Some(starting_id) = starting_id {

                    Range::from(starting_id..)

                    Range::from(..)

                let mut entries = Vec::new();

                task_self.roots().transactions().scan(range, |entry| {

                    entries.push(entry);

                    entries.len() < result_limit

                })?;

                entries

                    .into_iter()

                    .map(|entry| {

                        if let Some(data) = entry.data() {

                            let changes = match pot::from_slice(data) {

                                Ok(changes) => changes,