use std::borrow::Borrow;

use std::convert::Infallible;

use std::marker::PhantomData;

use std::ops::{Deref, DerefMut};

use std::string::FromUtf8Error;

use std::sync::Arc;

use actionable::{Action, Identifier};

use arc_bytes::serde::Bytes;

use async_trait::async_trait;

use futures::future::BoxFuture;

use futures::{Future, FutureExt};

use serde::{Deserialize, Serialize};

use zeroize::Zeroize;

use crate::admin::{Role, User};

use crate::document::{

    CollectionDocument, CollectionHeader, Document, HasHeader, Header, OwnedDocument,

};

use crate::key::{ByteSource, IntoPrefixRange, Key, KeyEncoding, KeyKind, KeyVisitor};

use crate::permissions::Permissions;

use crate::schema::view::map::{MappedDocuments, ViewMappings as ViewMappingsCurrent};

use crate::schema::{

    self, MappedValue, Nameable, NamedReference, Schema, SchemaName, SchemaSummary,

    SerializedCollection,

};

use crate::{transaction, Error};

mod has_session;

mod lowlevel;

pub use self::has_session::HasSession;

pub use self::lowlevel::{AsyncLowLevelConnection, HasSchema, LowLevelConnection};

/// A connection to a database's [`Schema`](schema::Schema), giving access to

/// [`Collection`s](crate::schema::Collection) and

/// [`Views`s](crate::schema::View). This trait is not safe to use within async

/// contexts and will block the current thread. For async access, use

/// [`AsyncConnection`].

pub trait Connection: LowLevelConnection + Sized + Send + Sync {

    /// The [`StorageConnection`] type that is paired with this type.

    type Storage: StorageConnection<Database = Self>;

    /// Returns the [`StorageConnection`] implementor that this database belongs to.

    fn storage(&self) -> Self::Storage;

    /// Accesses a collection for the connected [`Schema`](schema::Schema).

    /// Accesses a [`schema::View`] from this connection.

    /// Lists [executed transactions](transaction::Executed) from this

    /// [`Schema`](schema::Schema). By default, a maximum of 1000 entries will

    /// be returned, but that limit can be overridden by setting `result_limit`.

    /// A hard limit of 100,000 results will be returned. To begin listing after

    /// another known `transaction_id`, pass `transaction_id + 1` into

    /// `starting_id`.

    fn list_executed_transactions(

        &self,

        starting_id: Option<u64>,

        result_limit: Option<u32>,

    ) -> Result<Vec<transaction::Executed>, Error>;

    /// Fetches the last transaction id that has been committed, if any.

    fn last_transaction_id(&self) -> Result<Option<u64>, Error>;

    /// Compacts the entire database to reclaim unused disk space.

    ///

    /// This process is done by writing data to a new file and swapping the file

    /// once the process completes. This ensures that if a hardware failure,

    /// power outage, or crash occurs that the original collection data is left

    /// untouched.

    ///

    /// ## Errors

    ///

    /// * [`Error::Other`]: an error occurred while compacting the database.

    fn compact(&self) -> Result<(), crate::Error>;

    /// Compacts the collection to reclaim unused disk space.

    ///

    /// This process is done by writing data to a new file and swapping the file

    /// once the process completes. This ensures that if a hardware failure,

    /// power outage, or crash occurs that the original collection data is left

    /// untouched.

    ///

    /// ## Errors

    ///

    /// * [`Error::CollectionNotFound`]: database `name` does not exist.

    /// * [`Error::Other`]: an error occurred while compacting the database.

    /// Compacts the key value store to reclaim unused disk space.

    ///

    /// This process is done by writing data to a new file and swapping the file

    /// once the process completes. This ensures that if a hardware failure,

    /// power outage, or crash occurs that the original collection data is left

    /// untouched.

    ///

    /// ## Errors

    ///

    /// * [`Error::Other`]: an error occurred while compacting the database.

    fn compact_key_value_store(&self) -> Result<(), crate::Error>;

}

/// Interacts with a collection over a `Connection`.

///

/// These examples in this type use this basic collection definition:

///

/// ```rust

/// use bonsaidb_core::schema::Collection;

/// use bonsaidb_core::Error;

/// use serde::{Deserialize, Serialize};

///

/// #[derive(Debug, Serialize, Deserialize, Default, Collection)]

/// #[collection(name = "MyCollection")]

/// # #[collection(core = bonsaidb_core)]

/// pub struct MyCollection {

///     pub rank: u32,

///     pub score: f32,

/// }

/// ```

pub struct Collection<'a, Cn, Cl> {

    connection: &'a Cn,

    _phantom: PhantomData<Cl>, /* allows for extension traits to be written for collections of specific types */

}

impl<'a, Cn, Cl> Clone for Collection<'a, Cn, Cl> {

}

impl<'a, Cn, Cl> Collection<'a, Cn, Cl>

where

    Cn: Connection,

    Cl: schema::Collection,

{

    /// Creates a new instance using `connection`.

    /// Adds a new `Document<Cl>` with the contents `item`.

    ///

    /// ## Automatic ID Assignment

    ///

    /// This function calls [`SerializedCollection::natural_id()`] to try to

    /// retrieve a primary key value from `item`. If an id is returned, the item

    /// is inserted with that id. If an id is not returned, an id will be

    /// automatically assigned, if possible, by the storage backend, which uses the [`Key`]

    /// trait to assign ids.

    ///

    /// ```rust

    /// # bonsaidb_core::__doctest_prelude!();

    /// # use bonsaidb_core::connection::Connection;

    /// # fn test_fn<C: Connection>(db: &C) -> Result<(), Error> {

    /// let inserted_header = db

    ///     .collection::<MyCollection>()

    ///     .push(&MyCollection::default())?;

    /// println!(

    ///     "Inserted id {} with revision {}",

    ///     inserted_header.id, inserted_header.revision

    /// );

    /// # Ok(())

    /// # }

    /// ```

        } else {

        }

    /// Adds a new `Document<Cl>` with the `contents`.

    ///

    /// ## Automatic ID Assignment

    ///

    /// An id will be automatically assigned, if possible, by the storage backend, which uses

    /// the [`Key`] trait to assign ids.

    ///

    /// ```rust

    /// # bonsaidb_core::__doctest_prelude!();

    /// # use bonsaidb_core::connection::Connection;

    /// # fn test_fn<C: Connection>(db: &C) -> Result<(), Error> {

    /// let inserted_header = db.collection::<MyCollection>().push_bytes(vec![])?;

    /// println!(

    ///     "Inserted id {} with revision {}",

    ///     inserted_header.id, inserted_header.revision

    /// );

    /// # Ok(())

    /// # }

    /// ```

    /// Adds a new `Document<Cl>` with the given `id` and contents `item`.

    ///

    /// ```rust

    /// # bonsaidb_core::__doctest_prelude!();

    /// # use bonsaidb_core::connection::Connection;

    /// # fn test_fn<C: Connection>(db: &C) -> Result<(), Error> {

    /// let inserted_header = db

    ///     .collection::<MyCollection>()

    ///     .insert(&42, &MyCollection::default())?;

    /// println!(

    ///     "Inserted id {} with revision {}",

    ///     inserted_header.id, inserted_header.revision

    /// );

    /// # Ok(())

    /// # }

    /// ```

    /// Adds a new `Document<Cl>` with the the given `id` and `contents`.

    ///

    /// ```rust

    /// # bonsaidb_core::__doctest_prelude!();

    /// # use bonsaidb_core::connection::Connection;

    /// # fn test_fn<C: Connection>(db: &C) -> Result<(), Error> {

    /// let inserted_header = db.collection::<MyCollection>().insert_bytes(&42, vec![])?;

    /// println!(

    ///     "Inserted id {} with revision {}",

    ///     inserted_header.id, inserted_header.revision

    /// );

    /// # Ok(())

    /// # }

    /// ```

    /// Updates an existing document. Upon success, `doc.revision` will be

    /// updated with the new revision.

    ///

    /// ```rust

    /// # bonsaidb_core::__doctest_prelude!();

    /// # use bonsaidb_core::connection::Connection;

    /// # fn test_fn<C: Connection>(db: &C) -> Result<(), Error> {

    /// if let Some(mut document) = db.collection::<MyCollection>().get(&42)? {

    ///     // modify the document

    ///     db.collection::<MyCollection>().update(&mut document);

    ///     println!("Updated revision: {:?}", document.header.revision);

    /// }

    /// # Ok(())

    /// # }

    /// ```

    /// Overwrites an existing document, or inserts a new document. Upon success,

    /// `doc.revision` will be updated with the new revision information.

    ///

    /// ```rust

    /// # bonsaidb_core::__doctest_prelude!();

    /// # use bonsaidb_core::connection::Connection;

    /// # fn test_fn<C: Connection>(db: &C) -> Result<(), Error> {

    /// if let Some(mut document) = db.collection::<MyCollection>().get(&42)? {

    ///     // modify the document

    ///     db.collection::<MyCollection>().overwrite(&mut document);

    ///     println!("Updated revision: {:?}", document.header.revision);

    /// }

    /// # Ok(())

    /// # }

    /// ```

    /// Retrieves a `Document<Cl>` with `id` from the connection.

    ///

    /// ```rust

    /// # bonsaidb_core::__doctest_prelude!();

    /// # use bonsaidb_core::connection::Connection;

    /// # fn test_fn<C: Connection>(db: &C) -> Result<(), Error> {

    /// if let Some(doc) = db.collection::<MyCollection>().get(&42)? {

    ///     println!(

    ///         "Retrieved bytes {:?} with revision {}",

    ///         doc.contents, doc.header.revision

    ///     );

    ///     let deserialized = MyCollection::document_contents(&doc)?;

    ///     println!("Deserialized contents: {:?}", deserialized);

    /// }

    /// # Ok(())

    /// # }

    /// ```

    /// Retrieves all documents matching `ids`. Documents that are not found

    /// are not returned, but no error will be generated.

    ///

    /// ```rust

    /// # bonsaidb_core::__doctest_prelude!();

    /// # use bonsaidb_core::connection::Connection;

    /// # fn test_fn<C: Connection>(db: &C) -> Result<(), Error> {

    /// for doc in db.collection::<MyCollection>().get_multiple(&[42, 43])? {

    ///     println!("Retrieved #{} with bytes {:?}", doc.header.id, doc.contents);

    ///     let deserialized = MyCollection::document_contents(&doc)?;

    ///     println!("Deserialized contents: {:?}", deserialized);

    /// }

    /// # Ok(())

    /// # }

    /// ```

    /// Retrieves all documents matching the range of `ids`.

    ///

    /// ```rust

    /// # bonsaidb_core::__doctest_prelude!();

    /// # use bonsaidb_core::connection::Connection;

    /// # fn test_fn<C: Connection>(db: &C) -> Result<(), Error> {

    /// for doc in db

    ///     .collection::<MyCollection>()

    ///     .list(&42..)

    ///     .descending()

    ///     .limit(20)

    ///     .query()?

    /// {

    ///     println!("Retrieved #{} with bytes {:?}", doc.header.id, doc.contents);

    ///     let deserialized = MyCollection::document_contents(&doc)?;

    ///     println!("Deserialized contents: {:?}", deserialized);

    /// }

    /// # Ok(())

    /// # }

    /// ```

    /// Retrieves all documents with ids that start with `prefix`.

    ///

    /// ```rust

    /// use bonsaidb_core::connection::Connection;

    /// use bonsaidb_core::document::OwnedDocument;

    /// use bonsaidb_core::schema::{Collection, Schematic, SerializedCollection};

    /// use bonsaidb_core::Error;

    /// use serde::{Deserialize, Serialize};

    ///

    /// #[derive(Debug, Serialize, Deserialize, Default, Collection)]

    /// #[collection(name = "MyCollection", primary_key = String)]

    /// # #[collection(core = bonsaidb_core)]

    /// pub struct MyCollection;

    ///

    /// fn starts_with_a<C: Connection>(db: &C) -> Result<Vec<OwnedDocument>, Error> {

    ///     db.collection::<MyCollection>()

    ///         .list_with_prefix("a")

    ///         .query()

    /// }

    /// ```

    /// Retrieves all documents.

    ///

    /// ```rust

    /// # bonsaidb_core::__doctest_prelude!();

    /// # use bonsaidb_core::connection::Connection;

    /// # fn test_fn<C: Connection>(db: &C) -> Result<(), Error> {

    /// for doc in db.collection::<MyCollection>().all().query()? {

    ///     println!("Retrieved #{} with bytes {:?}", doc.header.id, doc.contents);

    ///     let deserialized = MyCollection::document_contents(&doc)?;

    ///     println!("Deserialized contents: {:?}", deserialized);

    /// }

    /// # Ok(())

    /// # }

    /// ```

    /// Removes a `Document` from the database.

    ///

    /// ```rust

    /// # bonsaidb_core::__doctest_prelude!();

    /// # use bonsaidb_core::connection::Connection;

    /// # fn test_fn<C: Connection>(db: &C) -> Result<(), Error> {

    /// if let Some(doc) = db.collection::<MyCollection>().get(&42)? {

    ///     db.collection::<MyCollection>().delete(&doc)?;

    /// }

    /// # Ok(())

    /// # }

    /// ```

}

/// Retrieves a list of documents from a collection. This structure also offers

/// functions to customize the options for the operation.

#[must_use]

pub struct List<'a, Cn, Cl, PrimaryKey>

where

    Cl: schema::Collection,

    PrimaryKey: PartialEq + ?Sized,

    Cl::PrimaryKey: Borrow<PrimaryKey> + PartialEq<PrimaryKey>,

{

    collection: MaybeOwned<'a, Collection<'a, Cn, Cl>>,

    range: RangeRef<'a, Cl::PrimaryKey, PrimaryKey>,

    sort: Sort,

    limit: Option<u32>,

}

impl<'a, Cn, Cl, PrimaryKey> List<'a, Cn, Cl, PrimaryKey>

where

    Cl: schema::Collection,

    Cn: Connection,

    PrimaryKey: KeyEncoding<Cl::PrimaryKey> + PartialEq + 'a + ?Sized,

    Cl::PrimaryKey: Borrow<PrimaryKey> + PartialEq<PrimaryKey>,

{

    /// Lists documents by id in ascending order.

    /// Lists documents by id in descending order.

    /// Sets the maximum number of results to return.

    /// Returns the number of documents contained within the range.

    ///

    /// Order and limit are ignored if they were set.

    ///

    /// ```rust

    /// # bonsaidb_core::__doctest_prelude!();

    /// # use bonsaidb_core::connection::Connection;

    /// # fn test_fn<C: Connection>(db: &C) -> Result<(), Error> {

    /// println!(

    ///     "Number of documents with id 42 or larger: {}",

    ///     db.collection::<MyCollection>().list(&42..).count()?

    /// );

    /// println!(

    ///     "Number of documents in MyCollection: {}",

    ///     db.collection::<MyCollection>().all().count()?

    /// );

    /// # Ok(())

    /// # }

    /// ```

    /// Returns the list of headers for documents contained within the range.

    ///

    /// ```rust

    /// # bonsaidb_core::__doctest_prelude!();

    /// # use bonsaidb_core::connection::Connection;

    /// # fn test_fn<C: Connection>(db: &C) -> Result<(), Error> {

    /// # tokio::runtime::Runtime::new().unwrap().block_on(async {

    /// println!(

    ///     "Headers with id 42 or larger: {:?}",

    ///     db.collection::<MyCollection>().list(&42..).headers()?

    /// );

    /// println!(

    ///     "Headers in MyCollection: {:?}",

    ///     db.collection::<MyCollection>().all().headers()?

    /// );

    /// # Ok(())

    /// # })

    /// # }

    /// ```

    /// Retrieves the matching documents.

    ///

    /// ```rust

    /// # bonsaidb_core::__doctest_prelude!();

    /// # use bonsaidb_core::connection::Connection;

    /// # fn test_fn<C: Connection>(db: &C) -> Result<(), Error> {

    /// for doc in db.collection::<MyCollection>().all().query()? {

    ///     println!("Retrieved #{} with bytes {:?}", doc.header.id, doc.contents);

    ///     let deserialized = MyCollection::document_contents(&doc)?;

    ///     println!("Deserialized contents: {:?}", deserialized);

    /// }

    /// # Ok(())

    /// # }

    /// ```

}

/// Parameters to query a [`schema::View`].

///

/// The examples for this type use this view definition:

///

/// ```rust

/// # mod collection {

/// # bonsaidb_core::__doctest_prelude!();

/// # }

/// # use collection::MyCollection;

/// use bonsaidb_core::define_basic_unique_mapped_view;

/// use bonsaidb_core::document::{CollectionDocument, Emit};

/// use bonsaidb_core::schema::{

///     CollectionMapReduce, DefaultViewSerialization, Name, ReduceResult, View, ViewMapResult,

///     ViewMappedValue, ViewSchema,

/// };

///

/// #[derive(Debug, Clone, View, ViewSchema)]

/// #[view(collection = MyCollection, key = u32, value = f32, name = "scores-by-rank")]

/// # #[view(core = bonsaidb_core)]

/// # #[view_schema(core = bonsaidb_core)]

/// pub struct ScoresByRank;

///

/// impl CollectionMapReduce for ScoresByRank {

///     fn map<'doc>(

///         &self,

///         document: CollectionDocument<<Self::View as View>::Collection>,

///     ) -> ViewMapResult<'doc, Self::View> {

///         document

///             .header

///             .emit_key_and_value(document.contents.rank, document.contents.score)

///     }

///

///     fn reduce(

///         &self,

///         mappings: &[ViewMappedValue<'_, Self::View>],

///         rereduce: bool,

///     ) -> ReduceResult<Self::View> {

///         if mappings.is_empty() {

///             Ok(0.)

///         } else {

///             Ok(mappings.iter().map(|map| map.value).sum::<f32>() / mappings.len() as f32)

///         }

///     }

/// }

/// ```

#[must_use]

pub struct View<'a, Cn, V: schema::SerializedView, Key>

where

    V::Key: Borrow<Key> + PartialEq<Key>,

    Key: PartialEq + ?Sized,

{

    connection: &'a Cn,

    /// Key filtering criteria.

    pub key: Option<QueryKey<'a, V::Key, Key>>,

    /// The view's data access policy. The default value is [`AccessPolicy::UpdateBefore`].

    pub access_policy: AccessPolicy,

    /// The sort order of the query.

    pub sort: Sort,

    /// The maximum number of results to return.

    pub limit: Option<u32>,

    _view: PhantomData<V>,

}

impl<'a, Cn, V, Key> View<'a, Cn, V, Key>

where

    V::Key: Borrow<Key> + PartialEq<Key>,

    V: schema::SerializedView,

    Cn: Connection,

    Key: KeyEncoding<V::Key> + PartialEq + ?Sized,

{

    /// Filters for entries in the view with `key`.

    ///

    /// ```rust

    /// # bonsaidb_core::__doctest_prelude!();

    /// # use bonsaidb_core::connection::Connection;

    /// # fn test_fn<C: Connection>(db: C) -> Result<(), Error> {

    /// // score is an f32 in this example

    /// for mapping in ScoresByRank::entries(&db).with_key(&42).query()? {

    ///     assert_eq!(mapping.key, 42);

    ///     println!("Rank {} has a score of {:3}", mapping.key, mapping.value);

    /// }

    /// # Ok(())

    /// # }

    /// ```

    #[allow(clippy::missing_const_for_fn)] // false positive, destructors

    /// Filters for entries in the view with `keys`.

    ///

    /// ```rust

    /// # bonsaidb_core::__doctest_prelude!();

    /// # use bonsaidb_core::connection::Connection;

    /// # fn test_fn<C: Connection>(db: C) -> Result<(), Error> {

    /// // score is an f32 in this example

    /// for mapping in ScoresByRank::entries(&db).with_keys(&[42, 43]).query()? {

    ///     println!("Rank {} has a score of {:3}", mapping.key, mapping.value);

    /// }

    /// # Ok(())

    /// # }

    /// ```

    /// Filters for entries in the view with the range `keys`.

    ///

    /// ```rust

    /// # bonsaidb_core::__doctest_prelude!();

    /// # use bonsaidb_core::connection::Connection;

    /// # fn test_fn<C: Connection>(db: C) -> Result<(), Error> {

    /// // score is an f32 in this example

    /// for mapping in ScoresByRank::entries(&db).with_key_range(42..).query()? {

    ///     assert!(mapping.key >= 42);

    ///     println!("Rank {} has a score of {:3}", mapping.key, mapping.value);

    /// }

    /// # Ok(())

    /// # }

    /// ```

    /// Filters for entries in the view with keys that begin with `prefix`.

    ///

    /// ```rust

    /// # bonsaidb_core::__doctest_prelude!();

    /// # use bonsaidb_core::connection::Connection;

    /// # fn test_fn<C: Connection>(db: C) -> Result<(), Error> {

    /// #[derive(View, Debug, Clone)]

    /// #[view(name = "by-name", key = String, collection = MyCollection)]

    /// # #[view(core = bonsaidb_core)]

    /// struct ByName;

    ///

    /// // score is an f32 in this example

    /// for mapping in ByName::entries(&db).with_key_prefix("a").query()? {

    ///     assert!(mapping.key.starts_with("a"));

    ///     println!("{} in document {:?}", mapping.key, mapping.source);

    /// }

    /// # Ok(())

    /// # }

    /// ```

    /// Sets the access policy for queries.

    ///

    /// ```rust

    /// # bonsaidb_core::__doctest_prelude!();

    /// # use bonsaidb_core::connection::Connection;

    /// # fn test_fn<C: Connection>(db: C) -> Result<(), Error> {

    /// // score is an f32 in this example

    /// for mapping in ScoresByRank::entries(&db)

    ///     .with_access_policy(AccessPolicy::UpdateAfter)

    ///     .query()?

    /// {

    ///     println!("Rank {} has a score of {:3}", mapping.key, mapping.value);

    /// }

    /// # Ok(())

    /// # }

    /// ```

    /// Returns the matching mappings in ascending key order. This is the

    /// default sorting behavior.

    ///

    /// When more than one mapping exists for a single key, all matching

    /// mappings are returned as a unique entry. The resulting mappings are

    /// sorted only by the key, and as such, the order of mappings with the same

    /// key is undefined.

    ///

    /// ```rust

    /// # bonsaidb_core::__doctest_prelude!();

    /// # use bonsaidb_core::connection::Connection;

    /// # fn test_fn<C: Connection>(db: C) -> Result<(), Error> {

    /// // score is an f32 in this example

    /// for mapping in ScoresByRank::entries(&db).ascending().query()? {

    ///     println!("Rank {} has a score of {:3}", mapping.key, mapping.value);

    /// }

    /// # Ok(())

    /// # }

    /// ```

    /// Returns the matching mappings in descending key order.

    ///

    /// When more than one mapping exists for a single key, all matching

    /// mappings are returned as a unique entry. The resulting mappings are

    /// sorted only by the key, and as such, the order of mappings with the same

    /// key is undefined.

    ///

    /// ```rust

    /// # bonsaidb_core::__doctest_prelude!();

    /// # use bonsaidb_core::connection::Connection;

    /// # fn test_fn<C: Connection>(db: C) -> Result<(), Error> {

    /// // score is an f32 in this example

    /// for mapping in ScoresByRank::entries(&db).descending().query()? {

    ///     println!("Rank {} has a score of {:3}", mapping.key, mapping.value);

    /// }

    /// # Ok(())

    /// # }

    /// ```

    /// Sets the maximum number of results to return.

    ///

    /// ```rust

    /// # bonsaidb_core::__doctest_prelude!();

    /// # use bonsaidb_core::connection::Connection;

    /// # fn test_fn<C: Connection>(db: C) -> Result<(), Error> {

    /// // score is an f32 in this example

    /// let mappings = ScoresByRank::entries(&db).limit(10).query()?;

    /// assert!(mappings.len() <= 10);

    /// # Ok(())

    /// # }

    /// ```

    /// Executes the query and retrieves the results.

    ///

    /// ```rust

    /// # bonsaidb_core::__doctest_prelude!();

    /// # use bonsaidb_core::connection::Connection;

    /// # fn test_fn<C: Connection>(db: C) -> Result<(), Error> {

    /// // score is an f32 in this example

    /// for mapping in ScoresByRank::entries(&db).query()? {

    ///     println!("Rank {} has a score of {:3}", mapping.key, mapping.value);

    /// }

    /// # Ok(())

    /// # }

    /// ```

    /// Executes the query and retrieves the results with the associated [`Document`s](crate::document::OwnedDocument).

    ///

    /// ```rust

    /// # bonsaidb_core::__doctest_prelude!();

    /// # use bonsaidb_core::connection::Connection;

    /// # fn test_fn<C: Connection>(db: C) -> Result<(), Error> {

    /// for mapping in &ScoresByRank::entries(&db)

    ///     .with_key_range(42..=44)

    ///     .query_with_docs()?

    /// {

    ///     println!(

    ///         "Mapping from #{} with rank: {} and score: {}. Document bytes: {:?}",

    ///         mapping.document.header.id, mapping.key, mapping.value, mapping.document.contents

    ///     );

    /// }

    /// # Ok(())

    /// # }

    /// ```

    /// Executes the query and retrieves the results with the associated [`CollectionDocument`s](crate::document::CollectionDocument).

    ///

    /// ```rust

    /// # bonsaidb_core::__doctest_prelude!();

    /// # use bonsaidb_core::connection::Connection;

    /// # fn test_fn<C: Connection>(db: C) -> Result<(), Error> {

    /// for mapping in &ScoresByRank::entries(&db)

    ///     .with_key_range(42..=44)

    ///     .query_with_collection_docs()?

    /// {

    ///     println!(

    ///         "Mapping from #{} with rank: {} and score: {}. Deserialized Contents: {:?}",

    ///         mapping.document.header.id, mapping.key, mapping.value, mapping.document.contents

    ///     );

    /// }

    /// # Ok(())

    /// # }

    /// ```

    /// Executes a reduce over the results of the query

    ///

    /// ```rust

    /// # bonsaidb_core::__doctest_prelude!();

    /// # use bonsaidb_core::connection::Connection;

    /// # fn test_fn<C: Connection>(db: C) -> Result<(), Error> {

    /// // score is an f32 in this example

    /// let score = ScoresByRank::entries(&db).reduce()?;

    /// println!("Average score: {:3}", score);

    /// # Ok(())

    /// # }

    /// ```

    /// Executes a reduce over the results of the query, grouping by key.

    ///

    /// ```rust

    /// # bonsaidb_core::__doctest_prelude!();

    /// # use bonsaidb_core::connection::Connection;

    /// # fn test_fn<C: Connection>(db: C) -> Result<(), Error> {

    /// // score is an f32 in this example

    /// for mapping in ScoresByRank::entries(&db).reduce_grouped()? {

    ///     println!(

    ///         "Rank {} has an average score of {:3}",

    ///         mapping.key, mapping.value

    ///     );

    /// }

    /// # Ok(())

    /// # }

    /// ```

    /// Deletes all of the associated documents that match this view query.

    ///

    /// ```rust

    /// # bonsaidb_core::__doctest_prelude!();

    /// # use bonsaidb_core::connection::Connection;

    /// # fn test_fn<C: Connection>(db: C) -> Result<(), Error> {

    /// ScoresByRank::entries(&db).delete_docs()?;

    /// # Ok(())

    /// # }

    /// ```

}

/// This type is the result of `reduce_grouped()`. It is a list of all matching

/// keys and the reduced value of all mapped entries for that key.

pub type GroupedReductions<V> =

    Vec<MappedValue<<V as schema::View>::Key, <V as schema::View>::Value>>;

/// A connection to a database's [`Schema`](schema::Schema), giving access to

/// [`Collection`s](crate::schema::Collection) and

/// [`Views`s](crate::schema::View). All functions on this trait are safe to use

/// in an asynchronous context.

#[async_trait]

pub trait AsyncConnection: AsyncLowLevelConnection + Sized + Send + Sync {

    /// The [`AsyncStorageConnection`] type that is paired with this type.

    type Storage: AsyncStorageConnection<Database = Self>;

    /// Returns the [`StorageConnection`] implementor that this database belongs

    /// to.

    fn storage(&self) -> Self::Storage;

    /// Accesses a collection for the connected [`Schema`](schema::Schema).

    /// Accesses a [`schema::View`] from this connection.

    /// Lists [executed transactions](transaction::Executed) from this [`Schema`](schema::Schema). By default, a maximum of

    /// 1000 entries will be returned, but that limit can be overridden by

    /// setting `result_limit`. A hard limit of 100,000 results will be

    /// returned. To begin listing after another known `transaction_id`, pass

    /// `transaction_id + 1` into `starting_id`.

    async fn list_executed_transactions(

        &self,

        starting_id: Option<u64>,

        result_limit: Option<u32>,

    ) -> Result<Vec<transaction::Executed>, Error>;

    /// Fetches the last transaction id that has been committed, if any.

    async fn last_transaction_id(&self) -> Result<Option<u64>, Error>;

    /// Compacts the entire database to reclaim unused disk space.

    ///

    /// This process is done by writing data to a new file and swapping the file

    /// once the process completes. This ensures that if a hardware failure,

    /// power outage, or crash occurs that the original collection data is left

    /// untouched.

    ///

    /// ## Errors

    ///

    /// * [`Error::Other`]: an error occurred while compacting the database.

    async fn compact(&self) -> Result<(), crate::Error>;

    /// Compacts the collection to reclaim unused disk space.

    ///

    /// This process is done by writing data to a new file and swapping the file

    /// once the process completes. This ensures that if a hardware failure,

    /// power outage, or crash occurs that the original collection data is left

    /// untouched.

    ///

    /// ## Errors

    ///

    /// * [`Error::CollectionNotFound`]: database `name` does not exist.

    /// * [`Error::Other`]: an error occurred while compacting the database.

    /// Compacts the key value store to reclaim unused disk space.

    ///

    /// This process is done by writing data to a new file and swapping the file

    /// once the process completes. This ensures that if a hardware failure,

    /// power outage, or crash occurs that the original collection data is left

    /// untouched.

    ///

    /// ## Errors

    ///

    /// * [`Error::Other`]: an error occurred while compacting the database.

    async fn compact_key_value_store(&self) -> Result<(), crate::Error>;

}

/// Interacts with a collection over a `Connection`.

///

/// These examples in this type use this basic collection definition:

///

/// ```rust

/// use bonsaidb_core::schema::Collection;

/// use bonsaidb_core::Error;

/// use serde::{Deserialize, Serialize};

///

/// #[derive(Debug, Serialize, Deserialize, Default, Collection)]

/// #[collection(name = "MyCollection")]

/// # #[collection(core = bonsaidb_core)]

/// pub struct MyCollection {

///     pub rank: u32,

///     pub score: f32,

/// }

/// ```

pub struct AsyncCollection<'a, Cn, Cl> {

    connection: &'a Cn,

    _phantom: PhantomData<Cl>, /* allows for extension traits to be written for collections of specific types */

}

impl<'a, Cn, Cl> Clone for AsyncCollection<'a, Cn, Cl> {

}

impl<'a, Cn, Cl> AsyncCollection<'a, Cn, Cl>

where

    Cn: AsyncConnection,

    Cl: schema::Collection,

{

    /// Creates a new instance using `connection`.

    /// Adds a new `Document<Cl>` with the contents `item`.

    ///

    /// ## Automatic ID Assignment

    ///

    /// This function calls [`SerializedCollection::natural_id()`] to try to

    /// retrieve a primary key value from `item`. If an id is returned, the item

    /// is inserted with that id. If an id is not returned, an id will be

    /// automatically assigned, if possible, by the storage backend, which uses the [`Key`]

    /// trait to assign ids.

    ///

    /// ```rust

    /// # bonsaidb_core::__doctest_prelude!();

    /// # use bonsaidb_core::connection::AsyncConnection;

    /// # fn test_fn<C: AsyncConnection>(db: &C) -> Result<(), Error> {

    /// # tokio::runtime::Runtime::new().unwrap().block_on(async {

    /// let inserted_header = db

    ///     .collection::<MyCollection>()

    ///     .push(&MyCollection::default())

    ///     .await?;

    /// println!(

    ///     "Inserted id {} with revision {}",

    ///     inserted_header.id, inserted_header.revision

    /// );

    /// # Ok(())

    /// # })

    /// # }

    /// ```

        } else {

        }

    /// Adds a new `Document<Cl>` with the `contents`.

    ///

    /// ## Automatic ID Assignment

    ///

    /// An id will be automatically assigned, if possible, by the storage backend, which uses

    /// the [`Key`] trait to assign ids.

    ///

    /// ```rust

    /// # bonsaidb_core::__doctest_prelude!();

    /// # use bonsaidb_core::connection::AsyncConnection;

    /// # fn test_fn<C: AsyncConnection>(db: &C) -> Result<(), Error> {

    /// # tokio::runtime::Runtime::new().unwrap().block_on(async {

    /// let inserted_header = db.collection::<MyCollection>().push_bytes(vec![]).await?;

    /// println!(

    ///     "Inserted id {} with revision {}",

    ///     inserted_header.id, inserted_header.revision

    /// );

    /// # Ok(())

    /// # })

    /// # }

    /// ```

    /// Adds a new `Document<Cl>` with the given `id` and contents `item`.

    ///

    /// ```rust

    /// # bonsaidb_core::__doctest_prelude!();

    /// # use bonsaidb_core::connection::AsyncConnection;

    /// # fn test_fn<C: AsyncConnection>(db: &C) -> Result<(), Error> {

    /// # tokio::runtime::Runtime::new().unwrap().block_on(async {

    /// let inserted_header = db

    ///     .collection::<MyCollection>()

    ///     .insert(&42, &MyCollection::default())

    ///     .await?;

    /// println!(

    ///     "Inserted id {} with revision {}",

    ///     inserted_header.id, inserted_header.revision

    /// );

    /// # Ok(())

    /// # })

    /// # }

    /// ```

    /// Adds a new `Document<Cl>` with the the given `id` and `contents`.

    ///

    /// ```rust

    /// # bonsaidb_core::__doctest_prelude!();

    /// # use bonsaidb_core::connection::AsyncConnection;

    /// # fn test_fn<C: AsyncConnection>(db: &C) -> Result<(), Error> {

    /// # tokio::runtime::Runtime::new().unwrap().block_on(async {

    /// let inserted_header = db

    ///     .collection::<MyCollection>()

    ///     .insert_bytes(&42, vec![])

    ///     .await?;

    /// println!(

    ///     "Inserted id {} with revision {}",

    ///     inserted_header.id, inserted_header.revision

    /// );

    /// # Ok(())

    /// # })

    /// # }

    /// ```

    /// Updates an existing document. Upon success, `doc.revision` will be

    /// updated with the new revision.

    ///

    /// ```rust

    /// # bonsaidb_core::__doctest_prelude!();

    /// # use bonsaidb_core::connection::AsyncConnection;

    /// # fn test_fn<C: AsyncConnection>(db: &C) -> Result<(), Error> {

    /// # tokio::runtime::Runtime::new().unwrap().block_on(async {

    /// if let Some(mut document) = db.collection::<MyCollection>().get(&42).await? {

    ///     // modify the document

    ///     db.collection::<MyCollection>().update(&mut document);

    ///     println!("Updated revision: {:?}", document.header.revision);

    /// }

    /// # Ok(())

    /// # })

    /// # }

    /// ```

    /// Overwrites an existing document, or inserts a new document. Upon success,

    /// `doc.revision` will be updated with the new revision information.

    ///

    /// ```rust

    /// # bonsaidb_core::__doctest_prelude!();

    /// # use bonsaidb_core::connection::AsyncConnection;

    /// # fn test_fn<C: AsyncConnection>(db: &C) -> Result<(), Error> {

    /// # tokio::runtime::Runtime::new().unwrap().block_on(async {

    /// if let Some(mut document) = db.collection::<MyCollection>().get(&42).await? {

    ///     // modify the document

    ///     db.collection::<MyCollection>().overwrite(&mut document);

    ///     println!("Updated revision: {:?}", document.header.revision);

    /// }

    /// # Ok(())

    /// # })

    /// # }

    /// ```

        doc.set_collection_header(