push notifications

src/domain/mod.rs

@@ -1,3 +1,4 @@
 pub mod agent;
 pub mod auth;
+pub mod notifications;
 pub mod tasks;

src/domain/notifications/mod.rs

@@ -0,0 +1 @@
+pub mod push;

src/domain/notifications/push.rs

@@ -0,0 +1,101 @@
+use crate::error::AppResult;
+use serde::{Deserialize, Serialize};
+use web_push::*;
+
+pub struct PushSender {
+    pub private_key: String,
+}
+
+#[derive(Serialize, Deserialize, Clone, Debug)]
+pub struct PushSubscription {
+    pub endpoint: String,
+    pub p256dh: String,
+    pub auth: String,
+}
+
+fn ensure_pem(input: &str) -> String {
+    let input = input.trim();
+    if input.contains("-----BEGIN") {
+        return input.to_string();
+    }
+
+    if input.starts_with("MHc") {
+        format!(
+            "-----BEGIN EC PRIVATE KEY-----\n{}\n-----END EC PRIVATE KEY-----",
+            input
+        )
+    } else {
+        format!(
+            "-----BEGIN PRIVATE KEY-----\n{}\n-----END PRIVATE KEY-----",
+            input
+        )
+    }
+}
+
+impl PushSender {
+    pub fn new(private_key_pem: &str) -> AppResult<Self> {
+        let pem = ensure_pem(private_key_pem);
+        // Validate key immediately to catch config errors early
+        let _ =
+            VapidSignatureBuilder::from_pem_no_sub(std::io::Cursor::new(&pem)).map_err(|e| {
+                crate::error::AppError::Internal(format!("Invalid VAPID private key: {}", e))
+            })?;
+
+        Ok(Self { private_key: pem })
+    }
+
+    pub fn get_public_key(&self) -> AppResult<Vec<u8>> {
+        let builder =
+            VapidSignatureBuilder::from_pem_no_sub(std::io::Cursor::new(&self.private_key))
+                .map_err(|e| crate::error::AppError::Internal(e.to_string()))?;
+        Ok(builder.get_public_key())
+    }
+
+    pub async fn send_notification(
+        &self,
+        subscription: &PushSubscription,
+        title: &str,
+        body: &str,
+    ) -> AppResult<()> {
+        let subscription_info = SubscriptionInfo::new(
+            subscription.endpoint.clone(),
+            subscription.p256dh.clone(),
+            subscription.auth.clone(),
+        );
+
+        let builder =
+            VapidSignatureBuilder::from_pem_no_sub(std::io::Cursor::new(&self.private_key))
+                .map_err(|e| crate::error::AppError::Internal(e.to_string()))?;
+
+        let vapid_signature = builder
+            .add_sub_info(&subscription_info)
+            .build()
+            .map_err(|e| crate::error::AppError::Internal(e.to_string()))?;
+
+        let mut builder = WebPushMessageBuilder::new(&subscription_info);
+
+        builder.set_vapid_signature(vapid_signature);
+
+        let payload = serde_json::to_vec(&serde_json::json!({
+            "title": title,
+            "body": body,
+        }))
+        .map_err(|e| crate::error::AppError::Internal(e.to_string()))?;
+
+        builder.set_payload(ContentEncoding::Aes128Gcm, &payload);
+
+        let message = builder
+            .build()
+            .map_err(|e| crate::error::AppError::Internal(e.to_string()))?;
+
+        let client = IsahcWebPushClient::new()
+            .map_err(|e| crate::error::AppError::Internal(e.to_string()))?;
+
+        client
+            .send(message)
+            .await
+            .map_err(|e| crate::error::AppError::Internal(e.to_string()))?;
+
+        Ok(())
+    }
+}

src/domain/tasks.rs

@@ -1,5 +1,5 @@
 use chrono::Utc;
-use sea_orm::{ActiveModelTrait, DatabaseConnection, EntityTrait, Set};
+use sea_orm::{ActiveModelTrait, ColumnTrait, DatabaseConnection, EntityTrait, QueryFilter, Set};
 use serde::{Deserialize, Serialize};
 use std::sync::Arc;
 use uuid::Uuid;
@@ -8,6 +8,7 @@ use crate::config::Config;
 use crate::domain::agent::Agent;
 use crate::entities::task::Entity as Task;
 use crate::entities::task_run::{self, Entity as TaskRun};
+use crate::entities::{push_subscription, task_subscription};
 use crate::scheduler::Scheduler;
 
 #[derive(Deserialize)]
@@ -119,7 +120,7 @@ pub async fn execute_agent_run(
     let mut run = run;
     run.logs = Set(logs.clone());
     run.answer = Set(answer.clone());
-    run.status = Set(status);
+    run.status = Set(status.clone());
 
     run.update(db)
         .await
@@ -132,6 +133,38 @@ pub async fn execute_agent_run(
             task_response.clone(),
         ));
 
+    // Send Push Notifications to subscribers
+    let subscriptions = task_subscription::Entity::find()
+        .filter(task_subscription::Column::TaskId.eq(task_id))
+        .all(db)
+        .await
+        .map_err(crate::error::AppError::Database)?;
+
+    for sub in subscriptions {
+        let push_subs = push_subscription::Entity::find()
+            .filter(push_subscription::Column::UserSub.eq(sub.user_sub))
+            .all(db)
+            .await
+            .map_err(crate::error::AppError::Database)?;
+
+        for push_sub in push_subs {
+            let sender = _scheduler.push_sender.clone();
+            let goal = task_response.goal.clone();
+            let status = status.clone();
+            let sub_data = crate::domain::notifications::push::PushSubscription {
+                endpoint: push_sub.endpoint,
+                p256dh: push_sub.p256dh,
+                auth: push_sub.auth,
+            };
+
+            tokio::spawn(async move {
+                let _ = sender
+                    .send_notification(&sub_data, &format!("Task Completed: {}", status), &goal)
+                    .await;
+            });
+        }
+    }
+
     Ok(task_response)
 }