use bsonrs::Document;

use crate::core::bsonc::Bsonc;
use crate::core::database::Database as CoreDatabase;

use crate::client::{Mongo, Client};
use crate::collection::Collection;
use crate::ReadPreference;
use crate::ReadConcern;
use crate::WriteConcern;
use crate::command::CommandOptions;
use crate::change_stream::{ChangeStream, WatchOptions};
use crate::aggregate::AggregateOptions;
use crate::cursor::Cursor;
use crate::error::Result;

pub mod options;

#[derive(Clone)]
pub struct DB {
    pub(crate) name: String,
    pub(crate) mongo: Mongo,
    pub(crate) read_preference: Option<ReadPreference>,
    pub(crate) read_concern: Option<ReadConcern>,
    pub(crate) write_concern: Option<WriteConcern>
}

impl DB {
    pub fn new(mongo: Mongo, name: &str) -> DB {
        DB::new_with_option(mongo, name, None, None, None)
    }

    pub fn new_with_option(
        mongo: Mongo,
        name: &str,
        read_preference: Option<ReadPreference>,
        read_concern: Option<ReadConcern>,
        write_concern: Option<WriteConcern>
    ) -> DB {
        is_send::<DB>();
        is_sync::<DB>();

        DB {
            name: name.to_string(),
            mongo,
            read_preference,
            read_concern,
            write_concern
        }
    }

    #[inline]
    pub fn name(&self) -> &str {
        &self.name
    }

    #[inline]
    pub fn acquire_client(&self) -> Client {
        self.mongo.acquire_client()
    }

    pub fn get_core_db(&self, client: &Client) -> CoreDatabase {
        let db = client.database(self.name());

        if let Some(read_prefs) = &self.read_preference {
            db.set_read_prefs(read_prefs.as_core());
        }

        if let Some(read_concern) = &self.read_concern {
            db.set_read_concern(&read_concern.as_core());
        }

        if let Some(write_concern) = &self.write_concern {
            db.set_write_concern(write_concern.as_core());
        }

        db
    }

    #[inline]
    pub fn collection(&self, name: &str) -> Collection {
        Collection::new(self.clone(), name)
    }

    /// This is a simplified interface to DB.command() that returns the first
    /// result document. The database’s read preference, read concern, and
    /// write concern are not applied to the command.
    ///
    /// # Example
    ///
    /// ```no_run
    /// use mongors::*;
    /// use bsonrs::doc;
    ///
    /// let client = Mongo::new("mongodb://localhost:27017/").unwrap();
    ///
    /// let cmd = doc!{
    ///     "dbStats": 1
    /// };
    ///
    /// let db = client.db("admin");
    ///
    /// let reply = db.command_simple(cmd, None);
    /// println!("{:?}", reply);
    /// ```
    pub fn command_simple(
        &self,
        command: Document,
        read_prefs: Option<&ReadPreference>
    ) ->  Result<Document> {
        let command = Bsonc::from_doc(&command)?;
        let read_prefs = read_prefs.map(|r| r.as_core());

        let client = self.acquire_client();
        let db = self.get_core_db(&client);

        let reply = db.command_simple(&command, read_prefs)?;
    
        Ok(reply.as_doc()?)
    }

    /// Execute a command on the server, interpreting opts according
    /// to the MongoDB server version.
    ///
    /// # Example
    ///
    /// ```no_run
    /// use mongors::*;
    /// use bsonrs::doc;
    ///
    /// let client = Mongo::new("mongodb://localhost:27017/").unwrap();
    ///
    /// let cmd = doc!{
    ///     "dbStats": 1
    /// };
    ///
    /// let db = client.db("admin");
    ///
    /// let reply = db.command(cmd, None, None);
    /// println!("{:?}", reply);
    /// ```
    pub fn command(
        &self,
        command: Document,
        read_prefs: Option<&ReadPreference>,
        opts: Option<CommandOptions>
    ) ->  Result<Document> {
        let command = Bsonc::from_doc(&command)?;
        let read_prefs = read_prefs.map(|r| r.as_core());

        let opts = match opts {
            Some(opts) => Some(Bsonc::from_doc(&opts.into())?),
            None => None
        };

        let client = self.acquire_client();
        let db = self.get_core_db(&client);

        let reply = db.command(&command, read_prefs, opts.as_ref())?;
    
        Ok(reply.as_doc()?)
    }

    pub fn watch(
        &self,
        pipeline: Document,
        opts: Option<WatchOptions>
    ) -> Result<ChangeStream> {
        let client = self.acquire_client();
        let db = self.get_core_db(&client);
        let pipeline = Bsonc::from_doc(&pipeline)?;

        let opts = match opts {
            Some(opts) => Some(Bsonc::from_doc(&opts.into())?),
            None => None
        };

        let core_charge_stream = db.watch(&pipeline, opts.as_ref());

        Ok(ChangeStream::new(core_charge_stream, client))
    }

    pub fn aggregate(
        &self,
        pipeline: Document,
        opts: Option<AggregateOptions>,
        read_prefs: Option<&ReadPreference>
    ) -> Result<Cursor> {
        let client = self.acquire_client();
        let db = self.get_core_db(&client);
        let pipeline = Bsonc::from_doc(&pipeline)?;
        
        let opts = match opts {
            Some(opts) => Some(Bsonc::from_doc(&opts.into())?),
            None => None
        };

        let read_prefs = read_prefs.map(|r| r.as_core());

        let core_cursor = db.aggregate(&pipeline, opts.as_ref(), read_prefs);

        Ok(Cursor::new(core_cursor, client))
    }
}

unsafe impl Send for DB {}
unsafe impl Sync for DB {}

fn is_send<T: Send>() {}
fn is_sync<T: Sync>() {}