Channel subscription, message sending, channel delete.

src/notifications/events.rs

@@ -36,6 +36,8 @@ pub enum ClientboundNotification {
         channels: Vec<Channel>
     },
 
+    Message(Message),
+
     /*MessageCreate {
         id: String,
         nonce: Option<String>,

src/notifications/subscriptions.rs

@@ -1,7 +1,10 @@
 use crate::database::*;
 
-use super::hive::get_hive;
+use futures::StreamExt;
+use mongodb::bson::doc;
 use hive_pubsub::PubSub;
+use super::hive::get_hive;
+use mongodb::options::FindOptions;
 
 pub async fn generate_subscriptions(user: &User) -> Result<(), String> {
     let hive = get_hive();
@@ -13,5 +16,37 @@ pub async fn generate_subscriptions(user: &User) -> Result<(), String> {
         }
     }
 
+    let mut cursor = get_collection("channels")
+        .find(
+            doc! {
+                "$or": [
+                    {
+                        "type": "SavedMessages",
+                        "user": &user.id
+                    },
+                    {
+                        "type": "DirectMessage",
+                        "recipients": &user.id,
+                        "active": true
+                    },
+                    {
+                        "type": "Group",
+                        "recipients": &user.id
+                    }
+                ]
+            },
+            FindOptions::builder()
+                .projection(doc! { "_id": 1 })
+                .build()
+        )
+        .await
+        .map_err(|_| "Failed to fetch channels.".to_string())?;
+    
+    while let Some(result) = cursor.next().await {
+        if let Ok(doc) = result {
+            hive.subscribe(user.id.clone(), doc.get_str("_id").unwrap().to_string())?;
+        }
+    }
+
     Ok(())
 }