//! MongoDB backend
//!
//! This is suitable for small applications,
//! but does have several drawbacks.
//!
//! All data is serialized into a JSON string
//! which is embedded into a Mongo document.

use crate::{Hive, PubSub};

use mongodb::options::{CursorType, FindOneOptions, FindOptions};
use serde::{de::DeserializeOwned, Serialize};
use futures::stream::StreamExt;
use mongodb::Collection;
use std::time::Duration;
use mongodb::bson::doc;
use serde_json::json;
use std::hash::Hash;
use std::thread;
use ulid::Ulid;

/// MongoDB pubsub
#[derive(Clone)]
pub struct MongodbPubSub<K: Clone + Eq + Hash, V: Clone + Eq + Hash + DeserializeOwned + Serialize, D: DeserializeOwned + Serialize> {
    hive: Hive<K, V, D>,
    collection: Collection,
    source: String
}

impl<K: Clone + Eq + Hash, V: Clone + Eq + Hash + DeserializeOwned + Serialize, D: DeserializeOwned + Serialize> MongodbPubSub<K, V, D> {
    pub fn new
        <F: 'static + Fn(Vec<K>, D) + Send>
        (callback: F, collection: Collection) -> MongodbPubSub<K, V, D> {
        MongodbPubSub {
            hive: Hive::new(callback),
            collection,
            source: Ulid::new().to_string()
        }
    }

    /// Start listening to the collection for incoming data.
    ///
    /// This code was lifted from the Revolt project.
    pub async fn listen(&self) -> Result<(), String> {
        // Start listening to the collection, we generate a Ulid as a reference.
        match self.collection.find(
            doc! {
                "_id": {
                    "$gt": Ulid::new().to_string()
                }
            },
            FindOptions::builder()
                .cursor_type(CursorType::TailableAwait)
                .no_cursor_timeout(true)
                .build(),
        )
        .await {
            Ok(mut cursor) => {
                loop {
                    while let Some(item) = cursor.next().await {
                        match item {
                            Ok(doc) => {
                                let source = doc.get_str("source");
                                if source.unwrap() == self.source {
                                    continue;
                                }

                                let topic = doc.get_str("topic");
                                let topic: V = serde_json::from_str(topic.unwrap()).map_err(|_| "")?;

                                let data = doc.get_str("data");
                                let data: D = serde_json::from_str(data.unwrap()).map_err(|_| "")?;

                                publish(&self, &topic, data).await?;
                            }
                            Err(error) =>
                                return Err(format!("Failed to fetch a document from collection! {}", error))
                        }
                    }
                }
            }
            Err(error) =>
                Err(format!("Failed to create a tailable cursor. {}", error))
        }
    }

    /// Set the id of this source.
    ///
    /// Used for testing.
    pub fn set_source(&mut self, source: String) {
        self.source = source;
    }
}

impl<K: Clone + Eq + Hash, V: Clone + Eq + Hash + DeserializeOwned + Serialize, D: DeserializeOwned + Serialize> PubSub<K, V, D> for MongodbPubSub<K, V, D> {
    fn drop_client(&self, id: &K) -> Result<(), String> {
        self.hive.drop_client(id)
    }

    fn drop_topic(&self, id: &V) -> Result<(), String> {
        self.hive.drop_topic(id)
    }

    fn subscribe(&self, client: K, topic: V) -> Result<(), String> {
        self.hive.subscribe(client, topic)
    }

    fn subscribe_multiple(&self, client: K, topics: Vec<V>) -> Result<(), String> {
        self.hive.subscribe_multiple(client, topics)
    }

    fn unsubscribe(&self, client: &K, topic: &V) -> Result<(), String> {
        self.hive.unsubscribe(client, topic)
    }

    fn unsubscribe_multiple(&self, client: &K, topics: &Vec<V>) -> Result<(), String> {
        self.hive.unsubscribe_multiple(client, topics)
    }

    fn publish(&self, _topic: &V, _data: D) -> Result<(), String> {
        unreachable!("Use backend::mongo::publish instead of backend::mongo::MongodbPubSub::publish.");
    }
}

pub async fn publish<K, V, D>(pubsub: &MongodbPubSub<K, V, D>, topic: &V, data: D) -> Result<(), String>
    where K: Clone + Eq + Hash, V: Clone + Eq + Hash + DeserializeOwned + Serialize, D: DeserializeOwned + Serialize
{
    pubsub.collection.insert_one(
        doc! {
            "_id": Ulid::new().to_string(),
            "topic": json!(topic).to_string(),
            "source": pubsub.source.to_string(),
            "data": json!(data).to_string()
        },
        None
    )
    .await
    .map_err(|_| "Failed to push to database.")?;

    pubsub.hive.publish(topic, data)
}
