use super::{Options, serde};
use crate::storage::{Error, object_store::ObjectStore};
use async_trait::async_trait;
use futures::{AsyncWriteExt, TryStreamExt};
use futures_util::AsyncReadExt;
use log::debug;
use mongodb::bson::{Bson, DateTime, Document, doc};
use mongodb::options::{ClientOptions, IndexOptions};
use mongodb::{Client, Collection, Database, IndexModel, results::UpdateResult};
use pbbson::Model;
use pbbson::bson::{Binary, spec::BinarySubtype};
use std::fs::File;
use std::io::{Read, Write};
use std::path::Path;
use std::time::SystemTime;
use strum_macros::AsRefStr;
use tempfile::NamedTempFile;
use url::Url;

#[derive(AsRefStr)]
enum Collections {
    #[strum(serialize = "fs.files")]
    FileInfos,
}

#[allow(unused)]
pub struct MongoDbObjectStore {
    pub(crate) db: Database,
}

impl MongoDbObjectStore {
    #[allow(unused)]
    pub async fn new(url: &Url, options: Options) -> Result<Self, Error> {
        let mut opts = ClientOptions::parse(url.to_string())
            .await
            .map_err(|e| Error::internal(e.to_string()))?;
        opts.app_name = options.app_name;
        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 { db };

        // Ensure indexes
        let _index_models = index_models();
        // TODO ensure_all_indexes().await?;

        Ok(this)
    }
}

#[async_trait]
impl ObjectStore for MongoDbObjectStore {
    async fn create(&self, metadata: Model, path: &Path) -> Result<Model, Error> {
        // Store file in the default GridFS bucket
        let mut file = File::open(path).map_err(|e| Error::internal(e.to_string()))?;
        let file_metadata = file.metadata().map_err(|e| Error::internal(e.to_string()))?;
        let mut buf = vec![0; file_metadata.len() as usize];
        file.read(&mut buf).map_err(|e| Error::internal(e.to_string()))?;
        let bucket = self.db.gridfs_bucket(None);
        let filename = metadata
            .get_str("filename")
            .map_err(|_e| Error::invalid_argument("A filename is required in metadata".to_string()))?;
        let mut upload_stream = bucket
            .open_upload_stream(filename)
            .await
            .map_err(|e| Error::internal(e.to_string()))?;
        upload_stream
            .write_all(&buf[..])
            .await
            .map_err(|e| Error::internal(e.to_string()))?;
        upload_stream
            .close()
            .await
            .map_err(|e| Error::internal(e.to_string()))?;
        let file_id = match upload_stream.id() {
            Bson::ObjectId(object_id) => serde::from_object_id(*object_id)?,
            _ => {
                return Err(Error::internal("Expected GridFS file id to be a BSON ObjectId"));
            }
        };

        // Load auto-populated fields like length and block size
        let mut final_metadata = self.find(&file_id).await?;

        // Transfer metadata
        let metadata = serde::transform_to_mongo(metadata)?;
        for (k, v) in metadata.iter() {
            if k == "id" || k == "length" || k == "filename" {
                continue;
            }
            final_metadata.insert(k, v);
        }

        // Finalize metadata and update
        let now = DateTime::from_system_time(SystemTime::now());
        final_metadata.set_datetime("createdAt", now);
        let hash = match metadata.get("md5") {
            Some(value) => value,
            None => &Bson::Binary(Binary {
                subtype: BinarySubtype::Md5,
                bytes: md5::compute(&buf).to_vec(),
            }),
        };
        final_metadata.insert("md5", hash);
        self.update(final_metadata).await
    }

    async fn delete(&self, id: &str, by_account_id: Option<String>) -> Result<(), Error> {
        let coll: Collection<Document> = self.db.collection(Collections::FileInfos.as_ref());
        let filter = doc! { "_id": serde::try_to_object_id(id)? };
        let now = SystemTime::now();
        let mut set = doc! {"deletedAt": DateTime::from_system_time(now)};
        if let Some(account_id) = by_account_id {
            set.insert("deletedBy", account_id);
        }
        let update = doc! {"$set": set};
        let res: UpdateResult = coll
            .update_one(filter, update)
            .await
            .map_err(|e| Error::internal(e.to_string()))?;
        if res.matched_count == 0 {
            return Err(Error::not_found("No such file"));
        }
        Ok(())
    }

    async fn download(&self, id: &str) -> Result<NamedTempFile, Error> {
        let bucket = self.db.gridfs_bucket(None);
        let id = serde::try_to_object_id(id).map_err(|e| Error::internal(e.to_string()))?;
        let mut download_stream = bucket
            .open_download_stream(Bson::ObjectId(id))
            .await
            .map_err(|e| Error::internal(e.to_string()))?;
        let mut buf = Vec::new();
        download_stream
            .read_to_end(&mut buf)
            .await
            .map_err(|e| Error::internal(e.to_string()))?;
        let mut file = NamedTempFile::new().map_err(|e| Error::internal(e.to_string()))?;
        file.write_all(&buf).map_err(|e| Error::internal(e.to_string()))?;
        file.flush().map_err(|e| Error::internal(e.to_string()))?;
        Ok(file)
    }

    async fn find(&self, id: &str) -> Result<Model, Error> {
        let coll: Collection<Document> = self.db.collection(Collections::FileInfos.as_ref());
        let filter = doc! {"_id": serde::try_to_object_id(id)?};
        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 file")),
            Some(document) => Ok(serde::transform_from_mongo(document.into())?),
        }
    }

    async fn find_many(&self, filter: Model, _ordering: Option<Model>) -> Result<Vec<Model>, Error> {
        let coll: Collection<Document> = self.db.collection(Collections::FileInfos.as_ref());
        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(doc) = cursor.try_next().await.map_err(|e| Error::internal(e.to_string()))? {
            models.push(serde::transform_from_mongo(doc.into())?);
        }
        Ok(models)
    }

    async fn update(&self, metadata: Model) -> Result<Model, Error> {
        let coll: Collection<Document> = self.db.collection(Collections::FileInfos.as_ref());
        let id = metadata.id()?;
        let now = DateTime::now();
        let mut metadata = serde::transform_to_mongo(metadata)?;
        metadata.insert("updatedAt", now);
        let filter = doc! {"_id": serde::try_to_object_id(&id)?};
        let _res: UpdateResult = coll
            .replace_one(filter, metadata.clone())
            .await
            .map_err(|e| Error::internal(e.to_string()))?;
        Ok(serde::transform_from_mongo(metadata)?)
    }
}

fn index_models() -> Vec<IndexModel> {
    vec![
        {
            IndexModel::builder()
                .keys({
                    doc! {"ownerAccountId": 1}
                })
                .options(Some(
                    IndexOptions::builder()
                        .name(Some("ownerAccountId_1".to_string()))
                        .build(),
                ))
                .build()
        },
        {
            IndexModel::builder()
                .keys({
                    doc! {"filename": 1, "length": 1}
                })
                .options(Some(
                    IndexOptions::builder()
                        .name(Some("filename_1_length_1".to_string()))
                        .build(),
                ))
                .build()
        },
    ]
}
