Add Pushover support. Fixes #3

notificohq · Dec 1, 2024 · 2cd1eb5 · 2cd1eb5
1 parent 235fa0f
commit 2cd1eb5
Show file tree

Hide file tree

Showing 11 changed files with 232 additions and 49 deletions.
diff --git a/.idea/notifico.iml b/.idea/notifico.iml
diff --git a/Cargo.toml b/Cargo.toml
@@ -18,7 +18,7 @@ members = [
     "notifico-project",
     "notifico-project/migration",
     "notifico-web",
-    "notifico-ingest",
+    "notifico-ingest", "transports/notifico-pushover",
 ]
 
 [workspace.dependencies]

diff --git a/notifico-core/src/pipeline/event.rs b/notifico-core/src/pipeline/event.rs
@@ -74,7 +74,7 @@ impl EventHandler {
                 event_id: msg.id,
             };
 
-            if context.recipient.is_none() {
+            if msg.recipients.is_empty() {
                 let task = serde_json::to_string(&PipelineTask { context }).unwrap();
                 self.task_tx.send(task).await.unwrap();
                 return Ok(());

diff --git a/notifico-web/assets/assets/index-BhrsKVmZ.js → notifico-web/assets/assets/index-Cj_sEHdQ.js b/notifico-web/assets/assets/index-BhrsKVmZ.js → notifico-web/assets/assets/index-Cj_sEHdQ.js
diff --git a/notifico-web/assets/index.html b/notifico-web/assets/index.html
@@ -105,7 +105,7 @@
             href="https://fonts.googleapis.com/css2?family=Roboto:wght@300;400;500;700&display=swap"
             rel="stylesheet"
         />
-      <script type="module" crossorigin src="./assets/index-BhrsKVmZ.js"></script>
+      <script type="module" crossorigin src="./assets/index-Cj_sEHdQ.js"></script>
     </head>
 
     <body>

diff --git a/notifico-worker/Cargo.toml b/notifico-worker/Cargo.toml
@@ -10,6 +10,7 @@ notifico-smtp = { path = "../transports/notifico-smtp" }
 notifico-whatsapp = { path = "../transports/notifico-whatsapp" }
 notifico-smpp = { path = "../transports/notifico-smpp" }
 notifico-slack = { path = "../transports/notifico-slack" }
+notifico-pushover = { path = "../transports/notifico-pushover" }
 
 notifico-template = { path = "../notifico-template" }
 notifico-subscription = { path = "../notifico-subscription" }

diff --git a/notifico-worker/src/amqp.rs b/notifico-worker/src/amqp.rs
@@ -2,17 +2,15 @@ use async_trait::async_trait;
 use fe2o3_amqp::connection::ConnectionHandle;
 use fe2o3_amqp::session::SessionHandle;
 use fe2o3_amqp::{Connection, Receiver, Sender, Session};
-use notifico_core::pipeline::event::EventHandler;
 use notifico_core::queue::{ReceiverChannel, SenderChannel};
-use std::collections::BTreeMap;
 use std::error::Error;
 use std::sync::Arc;
 use tokio::sync::Mutex;
 use tracing::info;
 use url::Url;
-use uuid::Uuid;
 
 pub struct AmqpClient {
+    #[allow(dead_code)]
     connection: ConnectionHandle<()>,
     session: SessionHandle<()>,
 }

diff --git a/notifico-worker/src/main.rs b/notifico-worker/src/main.rs
@@ -13,6 +13,7 @@ use notifico_core::pipeline::executor::PipelineExecutor;
 use notifico_core::queue::ReceiverChannel;
 use notifico_core::recorder::BaseRecorder;
 use notifico_dbpipeline::DbPipelineStorage;
+use notifico_pushover::PushoverPlugin;
 use notifico_slack::SlackPlugin;
 use notifico_smpp::SmppPlugin;
 use notifico_smtp::EmailPlugin;
@@ -131,6 +132,10 @@ async fn main() {
         credentials.clone(),
         recorder.clone(),
     )));
+    engine.add_plugin(Arc::new(PushoverPlugin::new(
+        credentials.clone(),
+        recorder.clone(),
+    )));
 
     let subman = Arc::new(SubscriptionManager::new(
         db_connection,

diff --git a/transports/notifico-pushover/Cargo.toml b/transports/notifico-pushover/Cargo.toml
@@ -0,0 +1,14 @@
+[package]
+name = "notifico-pushover"
+version = "0.1.0"
+edition = "2021"
+
+[dependencies]
+notifico-core = { path = "../../notifico-core" }
+reqwest = { workspace = true }
+async-trait = "0.1.83"
+serde = "1.0.215"
+serde_json = "1.0.133"
+serde_urlencoded = "0.7.1"
+thiserror = "2.0.3"
+url = "2.5.4"
diff --git a/transports/notifico-pushover/src/lib.rs b/transports/notifico-pushover/src/lib.rs
@@ -0,0 +1,154 @@
+mod step;
+
+use crate::step::{Step, STEPS};
+use async_trait::async_trait;
+use notifico_core::credentials::{CredentialStorage, TypedCredential};
+use notifico_core::engine::{EnginePlugin, PipelineContext, StepOutput};
+use notifico_core::error::EngineError;
+use notifico_core::recipient::TypedContact;
+use notifico_core::recorder::Recorder;
+use notifico_core::step::SerializedStep;
+use notifico_core::templater::RenderedTemplate;
+use serde::{Deserialize, Serialize};
+use std::borrow::Cow;
+use std::sync::Arc;
+use url::Url;
+
+#[derive(Debug, Serialize, Deserialize)]
+pub struct PushoverCredentials {
+    token: String,
+}
+
+impl TypedCredential for PushoverCredentials {
+    const CREDENTIAL_TYPE: &'static str = "pushover";
+}
+
+#[derive(Serialize, Deserialize)]
+struct PushoverMessageRequest {
+    token: String,
+    user: String,
+    message: String,
+
+    attachment_base64: Option<String>,
+    attachment_type: Option<String>,
+
+    device: Option<String>,
+    html: Option<u8>,
+    priority: Option<i8>,
+    sound: Option<String>,
+    timestamp: Option<u64>,
+    title: Option<String>,
+    ttl: Option<u64>,
+    url: Option<Url>,
+    url_title: Option<String>,
+}
+
+pub struct PushoverPlugin {
+    client: reqwest::Client,
+    credentials: Arc<dyn CredentialStorage>,
+    recorder: Arc<dyn Recorder>,
+}
+
+impl PushoverPlugin {
+    pub fn new(credentials: Arc<dyn CredentialStorage>, recorder: Arc<dyn Recorder>) -> Self {
+        Self {
+            client: reqwest::Client::new(),
+            credentials,
+            recorder,
+        }
+    }
+}
+
+#[async_trait]
+impl EnginePlugin for PushoverPlugin {
+    async fn execute_step(
+        &self,
+        context: &mut PipelineContext,
+        step: &SerializedStep,
+    ) -> Result<StepOutput, EngineError> {
+        let step: Step = step.clone().convert_step()?;
+
+        match step {
+            Step::Send { credential } => {
+                let credential: PushoverCredentials = self
+                    .credentials
+                    .get_typed_credential(context.project_id, &credential)
+                    .await?;
+
+                let contact: PushoverContact = context.get_contact()?;
+
+                for message in context.messages.iter().cloned() {
+                    let content: Message = message.content.try_into()?;
+                    let request = PushoverMessageRequest {
+                        token: credential.token.clone(),
+                        user: contact.user.clone(),
+                        message: content.text,
+                        attachment_base64: None,
+                        attachment_type: None,
+                        device: None,
+                        html: Some(1),
+                        priority: None,
+                        sound: None,
+                        timestamp: None,
+                        title: Some(content.title),
+                        ttl: None,
+                        url: None,
+                        url_title: None,
+                    };
+
+                    let result = self
+                        .client
+                        .post("https://api.pushover.net/1/messages.json")
+                        .body(serde_urlencoded::to_string(request).unwrap_or_default())
+                        .send()
+                        .await;
+
+                    match result {
+                        Ok(_) => self.recorder.record_message_sent(
+                            context.event_id,
+                            context.notification_id,
+                            message.id,
+                        ),
+                        Err(e) => self.recorder.record_message_failed(
+                            context.event_id,
+                            context.notification_id,
+                            message.id,
+                            &e.to_string(),
+                        ),
+                    }
+                }
+                Ok(StepOutput::Continue)
+            }
+        }
+    }
+
+    fn steps(&self) -> Vec<Cow<'static, str>> {
+        STEPS.iter().map(|&s| s.into()).collect()
+    }
+}
+
+#[derive(Serialize, Deserialize, Clone)]
+struct PushoverContact {
+    user: String,
+}
+
+impl TypedContact for PushoverContact {
+    const CONTACT_TYPE: &'static str = "pushover";
+}
+
+#[derive(Serialize, Deserialize, Clone)]
+pub struct Message {
+    pub text: String,
+    pub title: String,
+}
+
+impl TryFrom<RenderedTemplate> for Message {
+    type Error = EngineError;
+
+    fn try_from(value: RenderedTemplate) -> Result<Self, Self::Error> {
+        Ok(Self {
+            text: value.get("text")?.to_string(),
+            title: value.get("title")?.to_string(),
+        })
+    }
+}
diff --git a/transports/notifico-pushover/src/step.rs b/transports/notifico-pushover/src/step.rs
@@ -0,0 +1,10 @@
+use serde::{Deserialize, Serialize};
+
+#[derive(Serialize, Deserialize)]
+#[serde(tag = "step")]
+pub enum Step {
+    #[serde(rename = "pushover.send")]
+    Send { credential: String },
+}
+
+pub(crate) const STEPS: &[&str] = &["pushover.send"];