feat(task_queue): web push task

closes #113

src/database/entities/message.rs

@@ -218,6 +218,28 @@ impl Message {
         // if (channel => DM | Group) | mentions {
         //  spawn task_queue ( web push )
         // }
+        let mut target_ids = vec![];
+        match &channel {
+            Channel::DirectMessage { recipients, .. } | Channel::Group { recipients, .. } => {
+                for recipient in recipients {
+                    if !is_online(recipient) {
+                        target_ids.push(recipient.clone());
+                    }
+                }
+            }
+            Channel::TextChannel { .. } => {
+                if let Some(mentions) = &self.mentions {
+                    target_ids.append(&mut mentions.clone());
+                }
+            }
+            _ => {}
+        }
+
+        if target_ids.len() > 0 {
+            if let Ok(payload) = serde_json::to_string(&PushNotification::new(self, &channel).await) {
+                crate::task_queue::task_web_push::queue(target_ids, payload).await;
+            }
+        }
 
         /*
 
@@ -250,79 +272,7 @@ impl Message {
                     })?;*/
                     .unwrap();
             });
-        }
-
-        let mentions = self.mentions.clone();
-
-        /*
-           Web Push Test Code
-        */
-        let c_clone = channel.clone();
-        async_std::task::spawn(async move {
-            // Find all offline users.
-            let mut target_ids = vec![];
-            match &c_clone {
-                Channel::DirectMessage { recipients, .. } | Channel::Group { recipients, .. } => {
-                    for recipient in recipients {
-                        if !is_online(recipient) {
-                            target_ids.push(recipient.clone());
-                        }
-                    }
-                }
-                Channel::TextChannel { .. } => {
-                    if let Some(mut mentions) = mentions {
-                        target_ids.append(&mut mentions);
-                    }
-                }
-                _ => {}
-            }
-
-            // Fetch their corresponding sessions.
-            if target_ids.len() > 0 {
-                if let Ok(mut cursor) = Session::find(
-                        &get_db(),
-                        doc! {
-                            "_id": {
-                                "$in": target_ids
-                            },
-                            "subscription": {
-                                "$exists": true
-                            }
-                        },
-                        None
-                    )
-                    .await {
-                    let enc = serde_json::to_string(&PushNotification::new(self, &c_clone).await).unwrap();
-                    let client = WebPushClient::new();
-                    let key =
-                        base64::decode_config(VAPID_PRIVATE_KEY.clone(), base64::URL_SAFE).unwrap();
-
-                    while let Some(Ok(session)) = cursor.next().await {
-                        if let Some(sub) = session.subscription {
-                            let subscription = SubscriptionInfo {
-                                endpoint: sub.endpoint,
-                                keys: SubscriptionKeys {
-                                    auth: sub.auth,
-                                    p256dh: sub.p256dh
-                                }
-                            };
-
-                            let mut builder = WebPushMessageBuilder::new(&subscription).unwrap();
-                            let sig_builder = VapidSignatureBuilder::from_pem(
-                                std::io::Cursor::new(&key),
-                                &subscription,
-                            )
-                            .unwrap();
-                            let signature = sig_builder.build().unwrap();
-                            builder.set_vapid_signature(signature);
-                            builder.set_payload(ContentEncoding::AesGcm, enc.as_bytes());
-                            let m = builder.build().unwrap();
-                            client.send(m).await.ok();
-                        }
-                    }
-                }
-            }
-        });*/
+        }*/
 
         Ok(())
     }

src/task_queue/mod.rs

@@ -1,7 +1,9 @@
 pub mod task_last_message_id;
 pub mod task_process_embeds;
+pub mod task_web_push;
 
 pub async fn start_queues() {
     async_std::task::spawn(task_last_message_id::run());
     async_std::task::spawn(task_process_embeds::run());
+    async_std::task::spawn(task_web_push::run());
 }

src/task_queue/task_web_push.rs

@@ -0,0 +1,67 @@
+// Queue Type: Linear
+use async_channel::{ Sender, Receiver, bounded };
+use mongodb::bson::{doc};
+use web_push::{ContentEncoding, SubscriptionInfo, SubscriptionKeys, VapidSignatureBuilder, WebPushClient, WebPushMessageBuilder};
+use rauth::entities::{Model, Session};
+use futures::StreamExt;
+
+use crate::util::variables::VAPID_PRIVATE_KEY;
+use crate::database::*;
+
+type Message = (Vec<String>, String);
+
+lazy_static! {
+    static ref CHANNEL: (Sender<Message>, Receiver<Message>) = bounded(100);
+}
+
+pub async fn queue(recipients: Vec<String>, payload: String) {
+    CHANNEL.0.send((recipients, payload)).await.ok();
+}
+
+pub async fn run() {
+    let client = WebPushClient::new();
+    let key = base64::decode_config(VAPID_PRIVATE_KEY.clone(), base64::URL_SAFE)
+        .expect("valid `VAPID_PRIVATE_KEY`");
+
+    while let Ok((recipients, payload)) = CHANNEL.1.recv().await {
+        if let Ok(mut cursor) = Session::find(
+            &get_db(),
+            doc! {
+                "_id": {
+                    "$in": recipients
+                },
+                "subscription": {
+                    "$exists": true
+                }
+            },
+            None
+        )
+        .await {
+            while let Some(Ok(session)) = cursor.next().await {
+                if let Some(sub) = session.subscription {
+                    let subscription = SubscriptionInfo {
+                        endpoint: sub.endpoint,
+                        keys: SubscriptionKeys {
+                            auth: sub.auth,
+                            p256dh: sub.p256dh
+                        }
+                    };
+
+                    let mut builder = WebPushMessageBuilder::new(&subscription).unwrap();
+                    let sig_builder = VapidSignatureBuilder::from_pem(
+                        std::io::Cursor::new(&key),
+                        &subscription,
+                    )
+                    .unwrap();
+
+                    let signature = sig_builder.build().unwrap();
+                    builder.set_vapid_signature(signature);
+                    builder.set_payload(ContentEncoding::AesGcm, payload.as_bytes());
+
+                    let msg = builder.build().unwrap();
+                    client.send(msg).await.ok();
+                }
+            }
+        }
+    }
+}