pub(crate) mod serde;

use crate::storage::associations::{BelongsTo, HasMany};
use crate::storage::config_store::ConfigStore;
use crate::storage::Error;
use async_trait::async_trait;
use bson::{doc, Bson, DateTime};
use futures::TryStreamExt;
use log::{debug, error};
use mongodb::bson::Document;
use mongodb::options::{ReadConcern, WriteConcern};
use mongodb::results::{InsertOneResult, UpdateResult};
use mongodb::{options::ClientOptions, Client, ClientSession, Collection, Database, IndexModel};
use pbbson::Model;
use serde::{now_as_datetime, transform_from_mongo, try_to_object_id};
use std::collections::HashMap;
use url::Url;

#[allow(unused)]
pub struct MongodbConfigStore<B: Clone + Send + Sync + ToString> {
    belongs_tos_by_bucket: HashMap<String, Vec<BelongsTo<B>>>,
    db: Database,
    has_manys_by_bucket: HashMap<String, Vec<HasMany<B>>>,
    index_models_by_bucket: HashMap<String, Vec<IndexModel>>,
}

impl<B: Clone + Send + Sync + ToString> MongodbConfigStore<B> {
    #[allow(unused)]
    pub async fn new(
        url: &Url,
        app_name: &str,
        belongs_tos_by_bucket: HashMap<String, Vec<BelongsTo<B>>>,
        has_manys_by_bucket: HashMap<String, Vec<HasMany<B>>>,
        index_models_by_bucket: HashMap<String, Vec<IndexModel>>,
    ) -> Result<Self, Error> {
        let mut opts = ClientOptions::parse(url.to_string())
            .await
            .map_err(|e| Error::internal(e.to_string()))?;
        opts.app_name = Some(app_name.to_string());
        debug!(opts:?; "Using Mongo client options");
        let client = Client::with_options(opts).map_err(|e| Error::internal(e.to_string()))?;

        let database_name = {
            let mut s = url.path().to_string();
            if let Some(stripped) = s.strip_prefix("/") {
                s = stripped.to_string();
            }
            if let Some(found_at) = s.find("?") {
                s.truncate(found_at);
            }
            s.to_string()
        };
        debug!(database_name; "Using Mongo database");
        let db: Database = client.database(&database_name);

        // Validate the connection
        let _ = db
            .list_collection_names()
            .await
            .map_err(|e| Error::internal(e.to_string()))?;

        // Construct
        let this = Self {
            belongs_tos_by_bucket,
            db,
            has_manys_by_bucket,
            index_models_by_bucket,
        };

        // Ensure indexes
        this.ensure_all_indexes().await?;

        Ok(this)
    }
}

#[async_trait]
impl<B: Clone + Send + Sync + ToString> ConfigStore<B> for MongodbConfigStore<B> {
    async fn create(&self, bucket: B, model: Model) -> Result<Model, Error> {
        let mut session = self
            .db
            .client()
            .start_session()
            .await
            .map_err(|e| Error::internal(e.to_string()))?;
        session
            .start_transaction()
            .read_concern(ReadConcern::majority())
            .write_concern(WriteConcern::majority())
            .await
            .map_err(|e| Error::internal(e.to_string()))?;
        let coll: Collection<Document> = self.db.collection(&bucket.to_string());
        let now = now_as_datetime();

        // Set some defaults
        let mut model = model.clone();
        model.set_datetime("createdAt", now);
        if model.get("id").is_none() {
            model.insert("id", bson::oid::ObjectId::new());
        }

        // Work with a local copy from here on, where we replace id with _id
        let mut local_model = serde::transform_to_mongo(model.clone())?;
        let local_model_id = match model.get("id").unwrap() {
            Bson::ObjectId(object_id) => *object_id,
            Bson::String(id) => try_to_object_id(id)?,
            _ => return Err(Error::internal("Unsupported id type")),
        };
        local_model.insert("_id", local_model_id);

        // Store
        let _res: InsertOneResult = coll
            .insert_one(local_model.clone())
            .session(&mut session)
            .await
            .map_err(|e| Error::internal(e.to_string()))?;

        // Maintain belongs-to fk's
        for belongs_to in self
            .belongs_tos_by_bucket
            .get(&bucket.to_string())
            .cloned()
            .unwrap_or_default()
        {
            let local_value = local_model.get(&belongs_to.local);
            match local_value {
                Some(Bson::ObjectId(id)) => {
                    match belongs_to.inverse {
                        None => {
                            // TODO validate fk only
                        }
                        Some(inverse) => {
                            // Update foreign bucket to reference this
                            let add_to_set = doc! {&inverse: &local_model_id};
                            let set = doc! {"updatedAt": now/* TODO , "updatedByAccountId": &account_id*/};
                            let update = doc! {"$addToSet": add_to_set, "$set": set};
                            let remote_coll: Collection<Document> = self.db.collection(&belongs_to.remote.to_string());
                            let filter = doc! { "_id": id };
                            let res: UpdateResult = remote_coll
                                .update_one(filter, update)
                                .session(&mut session)
                                .await
                                .map_err(|e| Error::internal(e.to_string()))?;
                            if res.matched_count == 0 {
                                //error!("Could not find _ to add _ fk")
                            }
                        }
                    }
                }
                _ => {
                    error!("Unknown fk type: {local_value:?}");
                }
            }
        }

        // Maintain has-many fk's
        for has_many in self
            .has_manys_by_bucket
            .get(&bucket.to_string())
            .cloned()
            .unwrap_or_default()
        {
            let local_value = model.get(&has_many.local);
            match local_value {
                Some(Bson::Array(array)) => {
                    match has_many.inverse {
                        None => {
                            // TODO validate fk only
                        }
                        Some(inverse) => {
                            // Update foreign bucket to reference this
                            let remote_coll: Collection<Document> = self.db.collection(&has_many.remote.to_string());
                            for id in array.iter() {
                                let id = match id {
                                    Bson::ObjectId(object_id) => *object_id,
                                    Bson::String(id) => try_to_object_id(id)?,
                                    _ => return Err(Error::invalid_argument("Unsupport fk datatype")),
                                };
                                let set = doc! {&inverse: &local_model_id, "updatedAt": now/* TODO , "updatedByAccountId": &account_id*/};
                                let update = doc! {"$set": set};
                                let filter = doc! { "_id": id };
                                let res: UpdateResult = remote_coll
                                    .update_one(filter, update)
                                    .session(&mut session)
                                    .await
                                    .map_err(|e| Error::internal(e.to_string()))?;
                                if res.matched_count == 0 {
                                    //error!("Could not find _ to add _ fk")
                                }
                            }
                        }
                    }
                }
                _ => {
                    error!("Unknown fk type: {local_value:?}");
                }
            }
        }

        session
            .commit_transaction()
            .await
            .map_err(|e| Error::internal(e.to_string()))?;
        Ok(transform_from_mongo(model)?)
    }

    async fn delete(&self, bucket: B, id: &str, by_account_id: Option<String>) -> Result<(), Error> {
        let id = try_to_object_id(id)?;
        let mut session = self
            .db
            .client()
            .start_session()
            .await
            .map_err(|e| Error::internal(e.to_string()))?;
        session
            .start_transaction()
            .read_concern(ReadConcern::majority())
            .write_concern(WriteConcern::majority())
            .await
            .map_err(|e| Error::internal(e.to_string()))?;
        let coll: Collection<Document> = self.db.collection(&bucket.to_string());
        let filter = doc! { "_id": &id };
        let now = now_as_datetime();
        let mut set = doc! {"deletedAt": now};
        if let Some(account_id) = by_account_id {
            set.insert("deletedBy", account_id);
        }

        // Load existing
        let existing = self.find_in_session(&mut session, bucket.clone(), id).await?;

        // Soft delete
        let update = doc! {"$set": set};
        let res: UpdateResult = coll
            .update_one(filter, update)
            .session(&mut session)
            .await
            .map_err(|e| Error::internal(e.to_string()))?;
        if res.matched_count == 0 {
            return Err(Error::not_found("No such record"));
        }

        // Maintain belongs-to fk's
        for belongs_to in self
            .belongs_tos_by_bucket
            .get(&bucket.to_string())
            .cloned()
            .unwrap_or_default()
        {
            match belongs_to.inverse {
                None => continue,
                Some(inverse) => {
                    let local_value = existing.get(&belongs_to.local);
                    match local_value {
                        Some(Bson::ObjectId(id_)) => {
                            let pull = doc! {&inverse: id};
                            let set = doc! {"updatedAt": now/* TODO , "updatedByAccountId": &account_id*/};
                            let update = doc! {"$pull": pull, "$set": set};
                            let remote_coll: Collection<Document> = self.db.collection(&belongs_to.remote.to_string());
                            let filter = doc! { "_id": id_ };
                            let res: UpdateResult = remote_coll
                                .update_one(filter, update)
                                .session(&mut session)
                                .await
                                .map_err(|e| Error::internal(e.to_string()))?;
                            if res.matched_count == 0 {
                                //error!("Could not find _ to add _ fk")
                            }
                        }
                        _ => {
                            error!("Unknown fk type: {local_value:?}");
                        }
                    }
                }
            }
        }

        // Maintain has-many fk's
        for has_many in self
            .has_manys_by_bucket
            .get(&bucket.to_string())
            .cloned()
            .unwrap_or_default()
        {
            let local_value = existing.get(&has_many.local);
            match local_value {
                Some(Bson::Array(array)) => {
                    match has_many.inverse {
                        None => {
                            // TODO validate fk only
                        }
                        Some(inverse) => {
                            // Update foreign bucket to dereference this
                            let remote_coll: Collection<Document> = self.db.collection(&has_many.remote.to_string());
                            for id in array.iter() {
                                if let Bson::String(id_) = id {
                                    let set = doc! {&inverse: Bson::Null, "updatedAt": now/* TODO , "updatedByAccountId": &account_id*/};
                                    //let unset = doc! {&inverse: 1};
                                    let update = doc! {"$set": set};
                                    let filter = doc! { "_id": try_to_object_id(id_)? };
                                    let res: UpdateResult = remote_coll
                                        .update_one(filter, update)
                                        .session(&mut session)
                                        .await
                                        .map_err(|e| Error::internal(e.to_string()))?;
                                    if res.matched_count == 0 {
                                        //error!("Could not find _ to add _ fk")
                                    }
                                }
                            }
                        }
                    }
                }
                _ => {
                    error!("Unknown fk type: {local_value:?}");
                }
            }
        }

        session
            .commit_transaction()
            .await
            .map_err(|e| Error::internal(e.to_string()))?;
        Ok(())
    }

    async fn find(&self, bucket: B, id: &str) -> Result<Model, Error> {
        let coll: Collection<Document> = self.db.collection(&bucket.to_string());
        let filter = doc! {"_id": try_to_object_id(id)?, "deletedAt": None::<DateTime>};
        let maybe_document = coll
            .find_one(filter)
            .await
            .map_err(|e| Error::internal(e.to_string()))?;
        match maybe_document {
            None => Err(Error::not_found("No such record")),
            Some(document) => Ok(transform_from_mongo(document.into())?),
        }
    }

    async fn find_many(&self, bucket: B, filter: Model) -> Result<Vec<Model>, Error> {
        let coll: Collection<Document> = self.db.collection(&bucket.to_string());
        let filter = serde::transform_to_mongo(filter)?;
        let mut cursor = coll
            .find(filter.into())
            .await
            .map_err(|e| Error::internal(e.to_string()))?;
        let mut models = vec![];
        while let Some(document) = cursor.try_next().await.map_err(|e| Error::internal(e.to_string()))? {
            models.push(transform_from_mongo(document.into())?);
        }
        Ok(models)
    }

    async fn update(&self, bucket: B, model: Model) -> Result<Model, Error> {
        let mut session = self
            .db
            .client()
            .start_session()
            .await
            .map_err(|e| Error::internal(e.to_string()))?;
        session
            .start_transaction()
            .read_concern(ReadConcern::majority())
            .write_concern(WriteConcern::majority())
            .await
            .map_err(|e| Error::internal(e.to_string()))?;

        // Find existing, required to properly maintain fks
        let coll: Collection<Document> = self.db.collection(&bucket.to_string());
        let model_id = try_to_object_id(&model.id().map_err(|e| Error::internal(e.to_string()))?)?;
        let existing = self.find_in_session(&mut session, bucket.clone(), model_id).await?;

        let mut model = model.clone();
        let now = now_as_datetime();
        model.set_datetime("updatedAt", now);

        // Store the update
        let local_model = serde::transform_to_mongo(model.clone())?;
        let filter = doc! {"_id": &model_id};
        let _res: UpdateResult = coll
            .replace_one(filter, local_model)
            .session(&mut session)
            .await
            .map_err(|e| Error::internal(e.to_string()))?;

        // Maintain fk's of dereferenced belongs-tos
        let belongs_tos = self
            .belongs_tos_by_bucket
            .get(&bucket.to_string())
            .cloned()
            .unwrap_or_default();
        for belongs_to in belongs_tos.iter() {
            let existing_fk = existing.get(&belongs_to.local);
            let new_fk = model.get(&belongs_to.local);
            let new_fk = match new_fk.as_ref() {
                Some(Bson::String(s)) => Some(Bson::ObjectId(try_to_object_id(s)?)),
                _ => new_fk.cloned(),
            };
            match existing_fk {
                None => continue,
                Some(existing_fk_value) => {
                    if existing_fk == new_fk.as_ref() {
                        continue;
                    }
                    let existing_fk_object_id = match existing_fk_value.as_object_id() {
                        Some(existing_fk_object_id) => existing_fk_object_id,
                        None => continue,
                    };
                    let remote_model = self
                        .find_in_session(&mut session, belongs_to.remote.clone(), existing_fk_object_id)
                        .await?;
                    if let Some(inverse) = belongs_to.inverse.as_ref() {
                        let remote_fk_value = remote_model.get(inverse.clone());
                        if let Some(Bson::Array(array)) = remote_fk_value {
                            let array: Vec<_> = array
                                .iter()
                                .filter(|remote_fk| *remote_fk != &Bson::ObjectId(model_id))
                                .collect();
                            let mut remote_model = remote_model.clone();
                            remote_model.insert(inverse.clone(), array);
                            let filter = doc! {"_id": &existing_fk_object_id};
                            let remote_coll: Collection<Document> = self.db.collection(&belongs_to.remote.to_string());
                            let res: UpdateResult = remote_coll
                                .replace_one(filter, remote_model)
                                .session(&mut session)
                                .await
                                .map_err(|e| Error::internal(e.to_string()))?;
                            if res.modified_count > 0 {
                                debug!("Updated {} to remove fk", belongs_to.remote.to_string());
                            }
                        }
                    }
                }
            }
        }

        // Maintain fk's of newly referenced belongs-tos
        for belongs_to in belongs_tos.iter() {
            let existing_fk = existing.get(&belongs_to.local);
            let new_fk = model.get(&belongs_to.local);
            let new_fk = match new_fk.as_ref() {
                Some(Bson::String(s)) => Some(Bson::ObjectId(try_to_object_id(s)?)),
                _ => new_fk.cloned(),
            };
            match new_fk {
                None => continue,
                Some(ref new_fk_value) => {
                    if existing_fk == new_fk.as_ref() {
                        continue;
                    }
                    let new_fk_object_id = new_fk_value.as_object_id().unwrap();
                    let remote_model = self
                        .find_in_session(&mut session, belongs_to.remote.clone(), new_fk_object_id)
                        .await?;
                    if let Some(inverse) = belongs_to.inverse.as_ref() {
                        let remote_fk_value = remote_model.get(inverse.clone());
                        if let Some(Bson::Array(array)) = remote_fk_value {
                            let mut array = array.clone();
                            array.push(Bson::ObjectId(model_id));
                            let mut remote_model = remote_model.clone();
                            remote_model.insert(inverse.clone(), array);
                            let filter = doc! {"_id": &new_fk_object_id};
                            let remote_coll: Collection<Document> = self.db.collection(&belongs_to.remote.to_string());
                            let res: UpdateResult = remote_coll
                                .replace_one(filter, remote_model)
                                .session(&mut session)
                                .await
                                .map_err(|e| Error::internal(e.to_string()))?;
                            if res.modified_count > 0 {
                                debug!("Updated {} to add fk", belongs_to.remote.to_string());
                            }
                        }
                    }
                }
            }
        }

        // Maintain has-many fk's
        for has_many in self
            .has_manys_by_bucket
            .get(&bucket.to_string())
            .cloned()
            .unwrap_or_default()
        {
            let existing_local_value = existing.get(&has_many.local);
            let local_value = model.get(&has_many.local);
            match (existing_local_value, local_value) {
                (Some(Bson::Array(existing_array)), Some(Bson::Array(array))) => {
                    match has_many.inverse.clone() {
                        None => {
                            // TODO validate fk only
                        }
                        Some(inverse) => {
                            let remote_coll: Collection<Document> = self.db.collection(&has_many.remote.to_string());

                            // Maintain dereferenced fks
                            for existing_id in existing_array.iter() {
                                if array.contains(existing_id) {
                                    continue;
                                }
                                if let Bson::String(fk_object_id) = existing_id {
                                    let fk_object_id = try_to_object_id(fk_object_id)?;
                                    let set = doc! {&inverse: Bson::Null, "updatedAt": now/* TODO , "updatedByAccountId": &account_id*/};
                                    let update = doc! {"$set": set};
                                    let filter = doc! {"_id": &fk_object_id};
                                    let res: UpdateResult = remote_coll
                                        .update_one(filter, update)
                                        .session(&mut session)
                                        .await
                                        .map_err(|e| Error::internal(e.to_string()))?;
                                    if res.modified_count > 0 {
                                        debug!("Updated {} to remove fk", has_many.remote.to_string());
                                    }
                                }
                            }

                            // Maintain newly referenced fks
                            for id in array.iter() {
                                if existing_array.contains(id) {
                                    continue;
                                }
                                if let Bson::String(fk_object_id) = id {
                                    let fk_object_id = try_to_object_id(fk_object_id)?;
                                    let set = doc! {&inverse: &model_id, "updatedAt": now/* TODO , "updatedByAccountId": &account_id*/};
                                    let update = doc! {"$set": set};
                                    let filter = doc! {"_id": &fk_object_id};
                                    let res: UpdateResult = remote_coll
                                        .update_one(filter, update)
                                        .session(&mut session)
                                        .await
                                        .map_err(|e| Error::internal(e.to_string()))?;
                                    if res.modified_count > 0 {
                                        debug!("Updated {} to add fk", has_many.remote.to_string());
                                    }
                                }
                            }
                        }
                    }
                }
                _ => {
                    error!("Unknown fk type: {local_value:?}");
                }
            }
        }

        session
            .commit_transaction()
            .await
            .map_err(|e| Error::internal(e.to_string()))?;
        Ok(transform_from_mongo(model)?)
    }
}

impl<B: Clone + Send + Sync + ToString> MongodbConfigStore<B> {
    async fn ensure_all_indexes(&self) -> Result<(), Error> {
        for (bucket, index_models) in self.index_models_by_bucket.iter() {
            let coll: Collection<Document> = self.db.collection(bucket.as_ref());
            mongodb_ensure_index::ensure_indexes(&coll, index_models.clone())
                .await
                .map_err(|e| Error::internal(e.to_string()))?;
        }
        Ok(())
    }

    async fn find_in_session(
        &self,
        session: &mut ClientSession,
        bucket: B,
        id: bson::oid::ObjectId,
    ) -> Result<Model, Error> {
        let coll: Collection<Document> = self.db.collection(&bucket.to_string());
        let filter = doc! {"_id": id, "deletedAt": None::<DateTime>};
        let maybe_document = coll
            .find_one(filter)
            .session(session)
            .await
            .map_err(|e| Error::internal(e.to_string()))?;
        match maybe_document {
            None => Err(Error::not_found("No such record")),
            Some(document) => Ok(document.into()),
        }
    }
}
