1 files changed, 83 insertions, 0 deletions
diff --git a/src/push/publisher.rs b/src/push/publisher.rs
new file mode 100644
index 0000000..4092724
--- /dev/null
+++ b/src/push/publisher.rs
@@ -0,0 +1,83 @@
+use futures::future::join_all;
+use itertools::Itertools as _;
+use serde::Serialize;
+use web_push::{
+    ContentEncoding, IsahcWebPushClient, PartialVapidSignatureBuilder, SubscriptionInfo,
+    WebPushClient, WebPushError, WebPushMessage, WebPushMessageBuilder,
+};
+
+use crate::error::failed::{Failed, ResultExt as _};
+
+pub trait Publish {
+    fn publish<M>(
+        &self,
+        message: M,
+        signer: PartialVapidSignatureBuilder,
+        subscriptions: impl IntoIterator<Item = SubscriptionInfo> + Send,
+    ) -> impl Future<Output = Result<Vec<(SubscriptionInfo, WebPushError)>, Failed>> + Send
+    where
+        M: Serialize + Send + 'static;
+}
+
+#[derive(Clone)]
+pub struct Publisher {
+    client: IsahcWebPushClient,
+}
+
+impl Publisher {
+    pub fn new() -> Result<Self, WebPushError> {
+        let client = IsahcWebPushClient::new()?;
+        Ok(Self { client })
+    }
+
+    fn prepare_message(
+        payload: &[u8],
+        signer: &PartialVapidSignatureBuilder,
+        subscription: &SubscriptionInfo,
+    ) -> Result<WebPushMessage, Failed> {
+        let signature = signer
+            .clone()
+            .add_sub_info(subscription)
+            .build()
+            .fail("Failed to build VAPID signature")?;
+
+        let mut message = WebPushMessageBuilder::new(subscription);
+        message.set_payload(ContentEncoding::Aes128Gcm, payload);
+        message.set_vapid_signature(signature);
+        let message = message.build().fail("Failed to build push message")?;
+
+        Ok(message)
+    }
+}
+
+impl Publish for Publisher {
+    async fn publish<M>(
+        &self,
+        message: M,
+        signer: PartialVapidSignatureBuilder,
+        subscriptions: impl IntoIterator<Item = SubscriptionInfo> + Send,
+    ) -> Result<Vec<(SubscriptionInfo, WebPushError)>, Failed>
+    where
+        M: Serialize + Send + 'static,
+    {
+        let payload = serde_json::to_vec_pretty(&message)
+            .fail("Failed to encode web push message to JSON")?;
+
+        let messages: Vec<_> = subscriptions
+            .into_iter()
+            .map(|sub| Self::prepare_message(&payload, &signer, &sub).map(|message| (sub, message)))
+            .try_collect()?;
+
+        let deliveries = messages
+            .into_iter()
+            .map(async |(sub, message)| (sub, self.client.send(message).await));
+
+        let failures = join_all(deliveries)
+            .await
+            .into_iter()
+            .filter_map(|(sub, result)| result.err().map(|err| (sub, err)))
+            .collect();
+
+        Ok(failures)
+    }
+}