diff --git a/Cargo.lock b/Cargo.lock index bf2f121..650df82 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -130,18 +130,6 @@ version = "0.7.8" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "d3fb67a6e08acf24fdeccbac2cb6ac4305825bd1f117462e0e6f2f193345ad56" -[[package]] -name = "async-channel" -version = "2.5.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "924ed96dd52d1b75e9c1a3e6275715fd320f5f9439fb5a4a11fa51f4221158d2" -dependencies = [ - "concurrent-queue", - "event-listener-strategy", - "futures-core", - "pin-project-lite", -] - [[package]] name = "async-trait" version = "0.1.92" @@ -383,15 +371,6 @@ version = "1.12.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "fc652a48c352aef3ea3aed32080501cf3ef6ed5da78602a020c991775b0aff04" -[[package]] -name = "castaway" -version = "0.2.4" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "dec551ab6e7578819132c713a93c022a05d60159dc86e7a7050223577484c55a" -dependencies = [ - "rustversion", -] - [[package]] name = "cc" version = "1.4.4" @@ -498,15 +477,6 @@ dependencies = [ "memchr", ] -[[package]] -name = "concurrent-queue" -version = "2.5.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "4ca0197aee26d1ae37445ee532fefce43251d24cc7c166799f4d46817f1d3973" -dependencies = [ - "crossbeam-utils", -] - [[package]] name = "config" version = "0.1.0" @@ -706,37 +676,6 @@ dependencies = [ "cmov", ] -[[package]] -name = "curl" -version = "0.4.50" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "3a45ee8994e5307cb4c60cfc1c20bf7263ffb771ddc135c9f768a14bcbc15b09" -dependencies = [ - "curl-sys", - "libc", - "openssl-probe 0.1.6", - "openssl-sys", - "schannel", - "socket2", - "windows-sys 0.61.2", -] - -[[package]] -name = "curl-sys" -version = "0.4.90+curl-8.21.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "97799a0d220bfb3361e0fe4936966ff8c4b24d65c3f06dfc70d7b680b44e7897" -dependencies = [ - "cc", - "libc", - "libnghttp2-sys", - "libz-sys", - "openssl-sys", - "pkg-config", - "vcpkg", - "windows-sys 0.61.2", -] - [[package]] name = "der" version = "0.4.5" @@ -971,16 +910,6 @@ dependencies = [ "pin-project-lite", ] -[[package]] -name = "event-listener-strategy" -version = "0.5.4" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "8be9f3dfaaffdae2972880079a491a1a8bb7cbed0b8dd7a347f668b4150a3b93" -dependencies = [ - "event-listener", - "pin-project-lite", -] - [[package]] name = "event-publisher" version = "0.1.0" @@ -1002,12 +931,6 @@ dependencies = [ "zip", ] -[[package]] -name = "fastrand" -version = "2.5.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "da7c62ceae207dd37ea5b845da6a0696c799f85e97da1ab5b7910be3c1c80223" - [[package]] name = "ff" version = "0.13.1" @@ -1131,19 +1054,6 @@ version = "0.3.34" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "53c0fa8157de1303bfffdaa1cc2a673bfffb60102f76b0ef4441659124373fed" -[[package]] -name = "futures-lite" -version = "2.6.1" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "f78e10609fe0e0b3f4157ffab1876319b5b0db102a2c60dc4626306dc46b44ad" -dependencies = [ - "fastrand", - "futures-core", - "futures-io", - "parking", - "pin-project-lite", -] - [[package]] name = "futures-macro" version = "0.3.34" @@ -1292,12 +1202,6 @@ version = "0.5.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "2304e00983f87ffb38b55b444b5e3b60a884b5d30c0fca7d82fe33449bbe55ea" -[[package]] -name = "hermit-abi" -version = "0.5.2" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "fc0fef456e4baa96da950455cd02c081ca953b141298e41db3fc7e36b1da849c" - [[package]] name = "hex" version = "0.4.3" @@ -1690,32 +1594,6 @@ version = "2.12.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "6a756c3fac73139e83f14c2d742155dd2b78d3ee56597b419a0579b7bdd6dd78" -[[package]] -name = "isahc" -version = "1.8.3" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "fbce0b6b4f5c50b8e014e227d51ddf721558308566b4f6ba608abcea4d272cce" -dependencies = [ - "async-channel", - "castaway", - "crossbeam-utils", - "curl", - "curl-sys", - "encoding_rs", - "event-listener", - "futures-lite", - "http 0.2.12", - "log", - "mime", - "polling", - "slab", - "sluice", - "tracing", - "tracing-futures", - "url", - "waker-fn", -] - [[package]] name = "itertools" version = "0.15.0" @@ -1890,16 +1768,6 @@ version = "0.2.16" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "b6d2cec3eae94f9f509c767b45932f1ada8350c4bdb85af2fcab4a3c14807981" -[[package]] -name = "libnghttp2-sys" -version = "0.1.13+1.68.1" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "492e00167f1418c15648144f42bbfc63099806ecee9bf8d09a6353d6b4856b3c" -dependencies = [ - "cc", - "libc", -] - [[package]] name = "libsqlite3-sys" version = "0.30.1" @@ -1911,24 +1779,6 @@ dependencies = [ "vcpkg", ] -[[package]] -name = "libz-sys" -version = "1.1.29" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "85bc9657773828b90eeb625adff10eeac83cc21bbfd8e23a03eaa8a33c9e28d9" -dependencies = [ - "cc", - "libc", - "pkg-config", - "vcpkg", -] - -[[package]] -name = "linux-raw-sys" -version = "0.12.1" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "32a66949e030da00e8c7d4434b251670a91556f4144941d37452769c25d58a53" - [[package]] name = "litemap" version = "0.8.3" @@ -2223,12 +2073,6 @@ dependencies = [ "syn 2.0.119", ] -[[package]] -name = "openssl-probe" -version = "0.1.6" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "d05e27ee213611ffe7d6348b942e8f942b37114c00cc03cec254295a4a17852e" - [[package]] name = "openssl-probe" version = "0.2.1" @@ -2365,26 +2209,6 @@ dependencies = [ "siphasher", ] -[[package]] -name = "pin-project" -version = "1.1.13" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "2466b2336ed02bcdca6b294417127b90ec92038d1d5c4fbeac971a922e0e0924" -dependencies = [ - "pin-project-internal", -] - -[[package]] -name = "pin-project-internal" -version = "1.1.13" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "c96395f0a926bc13b1c17622aaddda1ecb55d49c8f1bf9777e4d877800a43f8b" -dependencies = [ - "proc-macro2", - "quote", - "syn 2.0.119", -] - [[package]] name = "pin-project-lite" version = "0.2.17" @@ -2428,20 +2252,6 @@ version = "0.3.34" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "f6b464fbc74e149a392436b17d523f769e057cb6877f6a5c4618bc6f11800548" -[[package]] -name = "polling" -version = "3.11.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "5d0e4f59085d47d8241c88ead0f274e8a0cb551f3625263c05eb8dd897c34218" -dependencies = [ - "cfg-if", - "concurrent-queue", - "hermit-abi", - "pin-project-lite", - "rustix", - "windows-sys 0.61.2", -] - [[package]] name = "polyval" version = "0.7.3" @@ -2770,19 +2580,6 @@ dependencies = [ "semver", ] -[[package]] -name = "rustix" -version = "1.1.4" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "b6fe4565b9518b83ef4f91bb47ce29620ca828bd32cb7e408f0062e9930ba190" -dependencies = [ - "bitflags", - "errno", - "libc", - "linux-raw-sys", - "windows-sys 0.61.2", -] - [[package]] name = "rustls" version = "0.23.43" @@ -2803,7 +2600,7 @@ version = "0.8.4" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "dab5152771c58876a2146916e53e35057e1a4dfa2b9df0f0305b07f611fdea4d" dependencies = [ - "openssl-probe 0.2.1", + "openssl-probe", "rustls-pki-types", "schannel", "security-framework", @@ -3182,17 +2979,6 @@ version = "0.4.12" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "0c790de23124f9ab44544d7ac05d60440adc586479ce501c1d6d7da3cd8c9cf5" -[[package]] -name = "sluice" -version = "0.6.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "160b744a45e8261307bcfe03c98e2f8274502207d534c9a64b675c4db1b6bd58" -dependencies = [ - "async-channel", - "futures-core", - "futures-io", -] - [[package]] name = "smallvec" version = "1.15.2" @@ -3868,16 +3654,6 @@ dependencies = [ "valuable", ] -[[package]] -name = "tracing-futures" -version = "0.2.5" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "97d095ae15e245a057c8e8451bab9b3ee1e1f68e9ba2b4fbc18d0ac5237835f2" -dependencies = [ - "pin-project", - "tracing", -] - [[package]] name = "tracing-log" version = "0.2.0" @@ -4071,12 +3847,6 @@ version = "0.9.5" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "0b928f33d975fc6ad9f86c8f283853ad26bdd5b10b7f1542aa2fa15e2289105a" -[[package]] -name = "waker-fn" -version = "1.2.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "317211a0dc0ceedd78fb2ca9a44aed3d7b9b26f81870d485c07122b4350673b7" - [[package]] name = "walkdir" version = "2.5.0" @@ -4189,9 +3959,7 @@ dependencies = [ "chrono", "ct-codecs", "ece", - "futures-lite", "http 0.2.12", - "isahc", "jwt-simple", "log", "pem 3.0.6", @@ -4209,6 +3977,7 @@ dependencies = [ "base64 0.23.1", "config", "domain", + "reqwest", "serde_json", "tracing", "web-push", diff --git a/Cargo.toml b/Cargo.toml index c35455d..5c9f498 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -56,6 +56,7 @@ argon2 = { version = "0.5", features = ["std"] } object_store = { version = "0.14", features = ["aws"] } bytes = "1" base64 = "0.23" +reqwest = { version = "0.13", default-features = false, features = ["rustls"] } utoipa = { version = "5", features = ["axum_extras", "chrono", "uuid"] } utoipa-scalar = { version = "0.3", features = ["axum"] } diff --git a/crates/adapters/web-push/Cargo.toml b/crates/adapters/web-push/Cargo.toml index acf4627..8652450 100644 --- a/crates/adapters/web-push/Cargo.toml +++ b/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 } diff --git a/crates/adapters/web-push/src/lib.rs b/crates/adapters/web-push/src/lib.rs index 28fd88d..bbcfd95 100644 --- a/crates/adapters/web-push/src/lib.rs +++ b/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, @@ -22,21 +23,12 @@ impl WebPushSender { config: &PushConfig, subscription_query: Arc, ) -> Result { - 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 { - 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 { + 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, 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::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 { + 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"); + } +} diff --git a/crates/server/src/main.rs b/crates/server/src/main.rs index 0d98006..f611a45 100644 --- a/crates/server/src/main.rs +++ b/crates/server/src/main.rs @@ -31,6 +31,8 @@ async fn run() -> Result<(), Box> { 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, +) { + 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()