use std::{

    borrow::Cow,

    collections::{HashMap, HashSet},

    fmt::{Debug, Display},

    fs::{self, File},

    io::{Read, Write},

    marker::PhantomData,

    path::{Path, PathBuf},

    sync::{Arc, Weak},

};

pub use bonsaidb_core::circulate::Relay;

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

use bonsaidb_core::document::KeyId;

use bonsaidb_core::{

    admin::{

        self,

        database::{self, ByName, Database as DatabaseRecord},

        user::User,

        Admin, PermissionGroup, Role, ADMIN_DATABASE_NAME,

    },

    circulate,

    connection::{

        self, Connection, HasSession, Identity, IdentityReference, LowLevelConnection, Session,

        SessionId, StorageConnection,

    },

    document::CollectionDocument,

    permissions::{

        bonsai::{

            bonsaidb_resource_name, database_resource_name, role_resource_name, user_resource_name,

            BonsaiAction, ServerAction,

        },

        Permissions,

    },

    schema::{Nameable, NamedCollection, Schema, SchemaName, Schematic},

};

#[cfg(feature = "password-hashing")]

use bonsaidb_core::{connection::Authentication, permissions::bonsai::AuthenticationMethod};

use itertools::Itertools;

use nebari::{

    io::{

        any::{AnyFile, AnyFileManager},

        FileManager,

    },

    ChunkCache, ThreadPool,

};

use parking_lot::{Mutex, RwLock};

use rand::{thread_rng, Rng};

#[cfg(feature = "compression")]

use crate::config::Compression;

#[cfg(feature = "encryption")]

use crate::vault::{self, LocalVaultKeyStorage, Vault};

use crate::{

    config::{KeyValuePersistence, StorageConfiguration},

    database::Context,

    tasks::{manager::Manager, TaskManager},

    Database, Error,

};

#[cfg(feature = "password-hashing")]

mod argon;

mod backup;

mod pubsub;

pub use backup::{AnyBackupLocation, BackupLocation};

/// A file-based, multi-database, multi-user database engine. This type blocks

/// the current thread when used. See [`AsyncStorage`](crate::AsyncStorage) for

/// this type's async counterpart.

///

/// ## Converting between Blocking and Async Types

///

/// [`AsyncStorage`](crate::AsyncStorage) and [`Storage`] can be converted to

/// and from each other using:

///

/// - [`AsyncStorage::into_blocking()`](crate::AsyncStorage::into_blocking)

/// - [`AsyncStorage::to_blocking()`](crate::AsyncStorage::to_blocking)

/// - [`AsyncStorage::as_blocking()`](crate::AsyncStorage::as_blocking)

/// - [`Storage::into_async()`]

/// - [`Storage::to_async()`]

/// - [`Storage::into_async_with_runtime()`]

/// - [`Storage::to_async_with_runtime()`]

///

/// ## Converting from `Database::open` to `Storage::open`

///

/// [`Database::open`](Database::open) is a simple method that uses `Storage` to

/// create a database named `default` with the schema provided. These two ways

/// of opening the database are the same:

///

/// ```rust

/// // `bonsaidb_core` is re-exported to `bonsaidb::core` or `bonsaidb_local::core`.

/// use bonsaidb_core::{connection::StorageConnection, schema::Schema};

/// // `bonsaidb_local` is re-exported to `bonsaidb::local` if using the omnibus crate.

/// use bonsaidb_local::{

///     config::{Builder, StorageConfiguration},

///     Database, Storage,

/// };

/// # fn open<MySchema: Schema>() -> anyhow::Result<()> {

/// // This creates a Storage instance, creates a database, and returns it.

/// let db = Database::open::<MySchema>(StorageConfiguration::new("my-db.bonsaidb"))?;

///

/// // This is the equivalent code being executed:

/// let storage =

///     Storage::open(StorageConfiguration::new("my-db.bonsaidb").with_schema::<MySchema>()?)?;

/// storage.create_database::<MySchema>("default", true)?;

/// let db = storage.database::<MySchema>("default")?;

/// #     Ok(())

/// # }

/// ```

///

/// ## Using multiple databases

///

/// This example shows how to use `Storage` to create and use multiple databases

/// with multiple schemas:

///

/// ```rust

/// use bonsaidb_core::{

///     connection::StorageConnection,

///     schema::{Collection, Schema},

/// };

/// use bonsaidb_local::{

///     config::{Builder, StorageConfiguration},

///     Storage,

/// };

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

///

/// #[derive(Debug, Schema)]

/// #[schema(name = "my-schema", collections = [BlogPost, Author])]

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

/// struct MySchema;

///

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

/// #[collection(name = "blog-posts")]

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

/// struct BlogPost {

///     pub title: String,

///     pub contents: String,

///     pub author_id: u64,

/// }

///

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

/// #[collection(name = "blog-posts")]

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

/// struct Author {

///     pub name: String,

/// }

///

/// # fn open() -> anyhow::Result<()> {

/// let storage = Storage::open(

///     StorageConfiguration::new("my-db.bonsaidb")

///         .with_schema::<BlogPost>()?

///         .with_schema::<MySchema>()?,

/// )?;

///

/// storage.create_database::<BlogPost>("ectons-blog", true)?;

/// let ectons_blog = storage.database::<BlogPost>("ectons-blog")?;

/// storage.create_database::<MySchema>("another-db", true)?;

/// let another_db = storage.database::<MySchema>("another-db")?;

/// # Ok(())

/// # }

/// ```

#[must_use]

pub struct Storage {

    pub(crate) instance: StorageInstance,

    pub(crate) authentication: Option<Arc<AuthenticatedSession>>,

    effective_session: Option<Arc<Session>>,

}

pub struct AuthenticatedSession {

    // TODO: client_data,

    storage: Weak<Data>,

    pub session: Session,

}

pub struct SessionSubscribers {

    pub subscribers: HashMap<u64, SessionSubscriber>,

    pub subscribers_by_session: HashMap<SessionId, HashSet<u64>>,

    pub last_id: u64,

}

impl SessionSubscribers {

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

        {

}

pub struct SessionSubscriber {

    pub session_id: Option<SessionId>,

    pub subscriber: circulate::Subscriber,

}

impl Drop for AuthenticatedSession {

    fn drop(&mut self) {

                // Deregister the session id once dropped.

}

struct AuthenticatedSessions {

    sessions: HashMap<SessionId, Arc<AuthenticatedSession>>,

    last_session_id: u64,

}

pub struct StorageInstance {

    data: Arc<Data>,

}

impl From<StorageInstance> for Storage {

}

struct Data {

    id: StorageId,

    path: PathBuf,

    parallelization: usize,

    threadpool: ThreadPool<AnyFile>,

    file_manager: AnyFileManager,

    pub(crate) tasks: TaskManager,

    schemas: RwLock<HashMap<SchemaName, Arc<dyn DatabaseOpener>>>,

    available_databases: RwLock<HashMap<String, SchemaName>>,

    open_roots: Mutex<HashMap<String, Context>>,

    // cfg check matches `Connection::authenticate`

    authenticated_permissions: Permissions,

    sessions: RwLock<AuthenticatedSessions>,

    pub(crate) subscribers: Arc<RwLock<SessionSubscribers>>,

    #[cfg(feature = "password-hashing")]

    argon: argon::Hasher,

    #[cfg(feature = "encryption")]

    pub(crate) vault: Arc<Vault>,

    #[cfg(feature = "encryption")]

    default_encryption_key: Option<KeyId>,

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

    tree_vault: Option<TreeVault>,

    pub(crate) key_value_persistence: KeyValuePersistence,

    chunk_cache: ChunkCache,

    pub(crate) check_view_integrity_on_database_open: bool,

    relay: Relay,

}

impl Storage {

    /// Creates or opens a multi-database [`Storage`] with its data stored in `directory`.

        } else {

        };

        #[cfg(feature = "encryption")]

                None => Arc::new(

                ),

            };

        };

    #[cfg(feature = "internal-apis")]

    #[doc(hidden)]

            // The configuraiton id override is not persisted to disk. This is

            // mostly to prevent someone from accidentally adding this

            // configuration, realizing it breaks things, and then wanting to

            // revert. This makes reverting to the old value easier.

        } else {

            // Load/Store a randomly generated id into a file. While the value

            // is numerical, the file contents are the ascii decimal, making it

            // easier for a human to view, and if needed, edit.

                // This value is important enought to not allow launching the

                // server if the file can't be read or contains unexpected data.

            } else {

            }

        }))

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

            Err(bonsaidb_core::Error::DatabaseNotFound(_)) => {

            }

        }

    /// Returns the unique id of the server.

    ///

    /// This value is set from the [`StorageConfiguration`] or randomly

    /// generated when creating a server. It shouldn't be changed after a server

    /// is in use, as doing can cause issues. For example, the vault that

    /// manages encrypted storage uses the server ID to store the vault key. If

    /// the server ID changes, the vault key storage will need to be updated

    /// with the new server ID.

    #[must_use]

    #[must_use]

    #[must_use]

    #[cfg(feature = "encryption")]

    #[must_use]

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

    #[must_use]

    #[cfg(feature = "encryption")]

    #[must_use]

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

    #[allow(clippy::unused_self)]

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

        None

    }

    /// Registers a schema for use within the server.

            )

        {

        } else {

        }

        } else {

        }

    /// Restricts an unauthenticated instance to having `effective_permissions`.

    /// Returns `None` if a session has already been established.

    #[must_use]

        } else {

        }

    /// Converts this instance into its blocking version, which is able to be

    /// used without async. The returned instance uses the current Tokio runtime

    /// handle to spawn blocking tasks.

    ///

    /// # Panics

    ///

    /// Panics if called outside the context of a Tokio runtime.

    #[cfg(feature = "async")]

    /// Converts this instance into its blocking version, which is able to be

    /// used without async. The returned instance uses the provided runtime

    /// handle to spawn blocking tasks.

    #[cfg(feature = "async")]

    /// Converts this instance into its blocking version, which is able to be

    /// used without async. The returned instance uses the current Tokio runtime

    /// handle to spawn blocking tasks.

    ///

    /// # Panics

    ///

    /// Panics if called outside the context of a Tokio runtime.

    #[cfg(feature = "async")]

    /// Converts this instance into its blocking version, which is able to be

    /// used without async. The returned instance uses the provided runtime

    /// handle to spawn blocking tasks.

    #[cfg(feature = "async")]

}

impl StorageInstance {

    #[cfg_attr(

        not(any(feature = "encryption", feature = "compression")),

        allow(unused_mut)

    )]

        } else {

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

        }

    /// Opens a database through a generic-free trait.

        // TODO switch to upgradable read now that we are on parking_lot

        };

        } else {

            // The schema was stored, the user is requesting the same schema,

            // but it isn't registerd with the storage currently.

        }

            }

            // TODO make this a generic not found with a name parameter.

        }

    #[cfg(feature = "password-hashing")]

            }

        }

        } else {

        }

        } else {

        }

}

pub trait DatabaseOpener: Send + Sync + Debug {

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

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

}

pub struct StorageSchemaOpener<DB: Schema> {

    schematic: Schematic,

    _phantom: PhantomData<DB>,

}

impl<DB> StorageSchemaOpener<DB>

where

    DB: Schema,

{

}

impl<DB> DatabaseOpener for StorageSchemaOpener<DB>

where

    DB: Schema,

{

}

impl HasSession for StorageInstance {

}

impl StorageConnection for StorageInstance {

    type Database = Database;

    type Authenticated = Storage;

    #[cfg_attr(

        feature = "tracing",

    )]

    fn create_database_with_schema(

        &self,

        name: &str,

        schema: SchemaName,

        only_if_needed: bool,

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

        Storage::validate_name(name)?;

        {

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

            if !schemas.contains_key(&schema) {

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

            }

        }

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

        let admin = self.admin();

        if !available_databases.contains_key(name) {

            admin

                .collection::<DatabaseRecord>()

                .push(&admin::Database {

                    name: name.to_string(),

                    schema: schema.clone(),

                })?;

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

        } else if !only_if_needed {

            return Err(bonsaidb_core::Error::DatabaseNameAlreadyTaken(

                name.to_string(),

            ));

        }

        Ok(())

    }

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

        let admin = self.admin();

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

        available_databases.remove(name);

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

        open_roots.remove(name);

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

        if database_folder.exists() {

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

            file_manager

                .delete_directory(&database_folder)

                .map_err(Error::Nebari)?;

        }

        if let Some(entry) = admin

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

            .with_key(name.to_ascii_lowercase())

            .query()?

            .first()

        {

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

            Ok(())

        } else {

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

        }

    }

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

        Ok(available_databases

            .iter()

            .collect())

    }

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

        Ok(available_databases.values().unique().cloned().collect())

    }

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

        let result = self

            .admin()

            .collection::<User>()

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

        Ok(result.id)

    }

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

        &self,

        user: U,

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

        let admin = self.admin();

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

        user.delete(&admin)?;

        Ok(())

    }

    #[cfg(feature = "password-hashing")]

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

        &self,

        user: U,

        password: bonsaidb_core::connection::SensitiveString,

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

        let admin = self.admin();

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

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

        user.update(&admin)

    }

    #[cfg(feature = "password-hashing")]

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

        &self,

        user: U,

        authentication: Authentication,

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

        let admin = self.admin();

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

        self.authenticate_inner(user, authentication, &admin)

            .map(Storage::from)

    }

            }

            }

        }

    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.update_user_with_named_id::<PermissionGroup, _, _, _>(

            user,

            permission_group,

        )

    }

    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.update_user_with_named_id::<PermissionGroup, _, _, _>(

            user,

            permission_group,

        )

    }

    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> {

    }

    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> {

    }

}

impl HasSession for Storage {

}

impl StorageConnection for Storage {

    type Database = Database;

    type Authenticated = Self;

    #[cfg_attr(

        feature = "tracing",

    )]

    fn create_database_with_schema(

        &self,

        name: &str,

        schema: SchemaName,

        only_if_needed: bool,

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

        self.check_permission(

            database_resource_name(name),

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

        )?;

        self.instance

            .create_database_with_schema(name, schema, only_if_needed)

    }

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

        self.check_permission(

            database_resource_name(name),

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

        )?;

        self.instance.delete_database(name)

    }

        self.check_permission(

            bonsaidb_resource_name(),

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

        )?;

        self.instance.list_databases()

    }

        self.check_permission(

            bonsaidb_resource_name(),

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

        )?;

        self.instance.list_available_schemas()

    }

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

        self.check_permission(

            bonsaidb_resource_name(),

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

        )?;

        self.instance.create_user(username)

    }

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

        &self,

        user: U,

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

        let admin = self.admin();

        let user = user.name()?;

        let user_id = user

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

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

        self.check_permission(

            user_resource_name(user_id),

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

        )?;

        self.instance.delete_user(user)

    }

    #[cfg(feature = "password-hashing")]

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

        &self,

        user: U,

        password: bonsaidb_core::connection::SensitiveString,

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

        let admin = self.admin();

        let user = user.name()?;

        let user_id = user

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

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

        self.check_permission(

            user_resource_name(user_id),

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

        )?;

        self.instance.set_user_password(user, password)

    }

    #[cfg(feature = "password-hashing")]

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

        &self,

        user: U,

        authentication: Authentication,

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

        let admin = self.admin();

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

        match &authentication {

            Authentication::Password(_) => {

                self.check_permission(

                    user_resource_name(user.header.id),

                    &BonsaiAction::Server(ServerAction::Authenticate(

                        AuthenticationMethod::PasswordHash,

                    )),

                )?;

            }

        }

        // TODO merge session permissions

        self.instance

            .authenticate_inner(user, authentication, &admin)

    }

            }

            }

        }

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

            .update_user_with_named_id::<PermissionGroup, _, _, _>(

                user,

                permission_group,

                |user, permission_group_id| {

            )

    }

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