Refactor web push notification handling to use reqwest client and improve session cleanup logic

crates/adapters/web-push/Cargo.toml

@@ -9,5 +9,6 @@ config = { workspace = true }
 async-trait = { workspace = true }
 tracing = { workspace = true }
 base64 = { workspace = true }
-web-push = "0.11"
+web-push = { version = "0.11", default-features = false }
 serde_json = { workspace = true }
+reqwest = { workspace = true }

crates/adapters/web-push/src/lib.rs

@@ -1,17 +1,18 @@
 use std::sync::Arc;
+use std::time::Duration;
 
 use web_push::{
-    ContentEncoding, IsahcWebPushClient, SubscriptionInfo, VapidSignatureBuilder, WebPushClient,
-    WebPushMessageBuilder,
+    ContentEncoding, SubscriptionInfo, VapidSignatureBuilder, WebPushMessage, WebPushMessageBuilder,
 };
 
 use config::PushConfig;
 use domain::errors::DomainError;
 use domain::ports::PushSubscriptionQueryPort;
+use domain::push::PushSubscription;
 use domain::user::UserId;
 
 pub struct WebPushSender {
-    client: IsahcWebPushClient,
+    client: reqwest::Client,
     vapid_private_key: String,
     vapid_subject: String,
     subscription_query: Arc<dyn PushSubscriptionQueryPort>,
@@ -22,21 +23,12 @@ impl WebPushSender {
         config: &PushConfig,
         subscription_query: Arc<dyn PushSubscriptionQueryPort>,
     ) -> Result<Self, DomainError> {
-        let private_key = config
-            .vapid_private_key
-            .as_deref()
-            .ok_or_else(|| DomainError::InvalidInput("vapid_private_key is required".into()))?;
+        let private_key = required_config(&config.vapid_private_key, "vapid_private_key")?;
+        let subject = required_config(&config.vapid_subject, "vapid_subject")?;
 
-        let subject = config
-            .vapid_subject
-            .as_deref()
-            .ok_or_else(|| DomainError::InvalidInput("vapid_subject is required".into()))?;
+        validate_vapid_key(private_key)?;
 
-        VapidSignatureBuilder::from_base64_no_sub(private_key)
-            .map_err(|e| DomainError::InvalidInput(format!("invalid VAPID key: {e}")))?;
-
-        let client = IsahcWebPushClient::new()
-            .map_err(|e| DomainError::InvalidInput(format!("failed to create push client: {e}")))?;
+        let client = build_http_client()?;
 
         Ok(Self {
             client,
@@ -47,22 +39,46 @@ impl WebPushSender {
     }
 
     pub fn public_key_base64(config: &PushConfig) -> Result<String, DomainError> {
-        let private_key = config
-            .vapid_private_key
-            .as_deref()
-            .ok_or_else(|| DomainError::InvalidInput("vapid_private_key is required".into()))?;
+        let private_key = required_config(&config.vapid_private_key, "vapid_private_key")?;
 
-        let sig_builder = VapidSignatureBuilder::from_base64_no_sub(private_key)
+        let partial = VapidSignatureBuilder::from_base64_no_sub(private_key)
             .map_err(|e| DomainError::InvalidInput(format!("invalid VAPID key: {e}")))?;
 
-        let public_key = sig_builder.get_public_key();
-        Ok(base64_url_encode(&public_key))
+        Ok(base64_url_encode(&partial.get_public_key()))
     }
-}
 
-fn base64_url_encode(input: &[u8]) -> String {
-    use base64::Engine;
-    base64::engine::general_purpose::URL_SAFE_NO_PAD.encode(input)
+    fn sign_for_subscription(
+        &self,
+        subscription_info: &SubscriptionInfo,
+    ) -> Result<web_push::VapidSignature, DomainError> {
+        let mut sig_builder =
+            VapidSignatureBuilder::from_base64(&self.vapid_private_key, subscription_info)
+                .map_err(|e| {
+                    DomainError::InvalidInput(format!("failed to build VAPID signature: {e}"))
+                })?;
+
+        sig_builder.add_claim("sub", &*self.vapid_subject);
+
+        sig_builder
+            .build()
+            .map_err(|e| DomainError::InvalidInput(format!("failed to sign push message: {e}")))
+    }
+
+    async fn deliver(&self, sub: &PushSubscription, payload: &str) -> Result<(), DomainError> {
+        let subscription_info = SubscriptionInfo::new(sub.endpoint(), sub.p256dh(), sub.auth());
+        let signature = self.sign_for_subscription(&subscription_info)?;
+        let message = build_push_message(&subscription_info, signature, payload)?;
+        let req = into_reqwest(&self.client, message);
+
+        match req.send().await {
+            Ok(response) => log_response(response, sub).await,
+            Err(e) => {
+                tracing::warn!(endpoint = sub.endpoint(), error = %e, "failed to send push");
+            }
+        }
+
+        Ok(())
+    }
 }
 
 #[async_trait::async_trait]
@@ -75,45 +91,98 @@ impl domain::ports::ReminderSenderPort for WebPushSender {
             return Ok(());
         }
 
-        let payload = serde_json::json!({
-            "title": "K-Mood",
-            "body": "How are you feeling right now?",
-            "url": "/"
-        });
-        let payload_str = payload.to_string();
+        let payload = reminder_payload();
 
         for sub in &subscriptions {
-            let subscription_info = SubscriptionInfo::new(sub.endpoint(), sub.p256dh(), sub.auth());
-
-            let mut sig_builder =
-                VapidSignatureBuilder::from_base64(&self.vapid_private_key, &subscription_info)
-                    .map_err(|e| {
-                        DomainError::InvalidInput(format!("failed to build VAPID signature: {e}"))
-                    })?;
-
-            sig_builder.add_claim("sub", &*self.vapid_subject);
-            let signature = sig_builder.build().map_err(|e| {
-                DomainError::InvalidInput(format!("failed to sign push message: {e}"))
-            })?;
-
-            let mut builder = WebPushMessageBuilder::new(&subscription_info);
-            builder.set_payload(ContentEncoding::Aes128Gcm, payload_str.as_bytes());
-            builder.set_vapid_signature(signature);
-
-            let message = builder.build().map_err(|e| {
-                DomainError::InvalidInput(format!("failed to build push message: {e}"))
-            })?;
-
-            match self.client.send(message).await {
-                Ok(_) => {
-                    tracing::info!(%user_id, endpoint = sub.endpoint(), "push notification sent");
-                }
-                Err(e) => {
-                    tracing::warn!(%user_id, endpoint = sub.endpoint(), error = %e, "failed to send push");
-                }
-            }
+            self.deliver(sub, &payload).await?;
         }
 
         Ok(())
     }
 }
+
+fn required_config<'a>(value: &'a Option<String>, name: &str) -> Result<&'a str, DomainError> {
+    value
+        .as_deref()
+        .ok_or_else(|| DomainError::InvalidInput(format!("{name} is required")))
+}
+
+fn validate_vapid_key(private_key: &str) -> Result<(), DomainError> {
+    VapidSignatureBuilder::from_base64_no_sub(private_key)
+        .map_err(|e| DomainError::InvalidInput(format!("invalid VAPID key: {e}")))?;
+    Ok(())
+}
+
+fn build_http_client() -> Result<reqwest::Client, DomainError> {
+    reqwest::Client::builder()
+        .pool_max_idle_per_host(2)
+        .pool_idle_timeout(Duration::from_secs(30))
+        .build()
+        .map_err(|e| DomainError::InvalidInput(format!("failed to create HTTP client: {e}")))
+}
+
+fn base64_url_encode(input: &[u8]) -> String {
+    use base64::Engine;
+    base64::engine::general_purpose::URL_SAFE_NO_PAD.encode(input)
+}
+
+fn reminder_payload() -> String {
+    serde_json::json!({
+        "title": "K-Mood",
+        "body": "How are you feeling right now?",
+        "url": "/"
+    })
+    .to_string()
+}
+
+fn build_push_message(
+    subscription_info: &SubscriptionInfo,
+    signature: web_push::VapidSignature,
+    payload: &str,
+) -> Result<WebPushMessage, DomainError> {
+    let mut builder = WebPushMessageBuilder::new(subscription_info);
+    builder.set_payload(ContentEncoding::Aes128Gcm, payload.as_bytes());
+    builder.set_vapid_signature(signature);
+
+    builder
+        .build()
+        .map_err(|e| DomainError::InvalidInput(format!("failed to build push message: {e}")))
+}
+
+fn into_reqwest(client: &reqwest::Client, message: WebPushMessage) -> reqwest::RequestBuilder {
+    let mut req = client
+        .post(message.endpoint.to_string())
+        .header("TTL", message.ttl.to_string());
+
+    if let Some(urgency) = message.urgency {
+        req = req.header("Urgency", urgency.to_string());
+    }
+
+    if let Some(topic) = message.topic {
+        req = req.header("Topic", topic);
+    }
+
+    if let Some(payload) = message.payload {
+        req = req
+            .header("Content-Encoding", payload.content_encoding.to_str())
+            .header("Content-Type", "application/octet-stream");
+
+        for (k, v) in payload.crypto_headers {
+            req = req.header(k, v);
+        }
+
+        req = req.body(payload.content);
+    }
+
+    req
+}
+
+async fn log_response(response: reqwest::Response, sub: &PushSubscription) {
+    let status = response.status();
+    if status.is_success() {
+        tracing::info!(endpoint = sub.endpoint(), "push notification sent");
+    } else {
+        let body = response.text().await.unwrap_or_default();
+        tracing::warn!(endpoint = sub.endpoint(), %status, body, "push endpoint rejected");
+    }
+}

crates/server/src/main.rs

@@ -31,6 +31,8 @@ async fn run() -> Result<(), Box<dyn std::error::Error>> {
         tracing::info!("push notifications enabled, reminder scheduler started");
     }
 
+    spawn_session_cleanup(context.state.refresh_session_command.clone());
+
     let router = http_axum::router::build_router(context.state);
 
     let addr = format!("{}:{}", config.server.host, config.server.port);
@@ -75,6 +77,25 @@ fn spawn_reminder_scheduler(
     });
 }
 
+fn spawn_session_cleanup(
+    refresh_session_command: std::sync::Arc<dyn domain::ports::RefreshSessionCommandPort>,
+) {
+    tokio::spawn(async move {
+        let mut interval = tokio::time::interval(std::time::Duration::from_secs(3600));
+        loop {
+            interval.tick().await;
+            match refresh_session_command.delete_expired().await {
+                Ok(deleted) => {
+                    if deleted > 0 {
+                        tracing::info!(deleted, "expired refresh sessions cleaned up");
+                    }
+                }
+                Err(e) => tracing::error!(error = %e, "refresh session cleanup failed"),
+            }
+        }
+    });
+}
+
 async fn shutdown_signal() {
     let ctrl_c = async {
         tokio::signal::ctrl_c()