From 0025bea23d4c1280746813139baef264e786ad5c Mon Sep 17 00:00:00 2001 From: Paul Date: Tue, 2 Nov 2021 12:50:35 +0000 Subject: [PATCH] feat(task_queue): web push task closes #113 --- src/database/entities/message.rs | 96 ++++++++------------------------ src/task_queue/mod.rs | 2 + src/task_queue/task_web_push.rs | 67 ++++++++++++++++++++++ 3 files changed, 92 insertions(+), 73 deletions(-) create mode 100644 src/task_queue/task_web_push.rs diff --git a/src/database/entities/message.rs b/src/database/entities/message.rs index 988e91c0..50fb417f 100644 --- a/src/database/entities/message.rs +++ b/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(()) } diff --git a/src/task_queue/mod.rs b/src/task_queue/mod.rs index 10f0b5f8..5f27dd65 100644 --- a/src/task_queue/mod.rs +++ b/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()); } diff --git a/src/task_queue/task_web_push.rs b/src/task_queue/task_web_push.rs new file mode 100644 index 00000000..8110b0a2 --- /dev/null +++ b/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); + +lazy_static! { + static ref CHANNEL: (Sender, Receiver) = bounded(100); +} + +pub async fn queue(recipients: Vec, 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(); + } + } + } + } +}