1use std::collections::HashMap;
2use std::sync::Arc;
3
4use serde::{Deserialize, Serialize};
5use tokio::sync::RwLock;
6use tracing::instrument;
7use web_push::{
8 ContentEncoding, IsahcWebPushClient, SubscriptionInfo, VapidSignatureBuilder, WebPushClient,
9 WebPushError, WebPushMessageBuilder,
10};
11
12use crate::auth::AuthState;
13use crate::repository::TaskRepository;
14use oxinbox_core::TaskStatus;
15
16#[derive(Debug, Clone, Serialize, Deserialize)]
17pub struct PushSubscription {
18 pub endpoint: String,
19 pub p256dh: String,
20 pub auth: String,
21}
22
23#[derive(Clone)]
24pub struct PushService {
25 pub subscriptions: Arc<RwLock<HashMap<i32, Vec<PushSubscription>>>>,
26 vapid_private_key: Option<String>,
27 vapid_contact: Option<String>,
28}
29
30impl PushService {
31 pub fn new() -> Self {
32 let vapid_private_key = std::env::var("VAPID_PRIVATE_KEY").ok();
33 let vapid_contact = std::env::var("VAPID_CONTACT").ok();
34 Self {
35 subscriptions: Arc::new(RwLock::new(HashMap::new())),
36 vapid_private_key,
37 vapid_contact,
38 }
39 }
40
41 pub const fn is_configured(&self) -> bool {
42 self.vapid_private_key.is_some()
43 }
44
45 #[instrument(skip(self))]
46 pub async fn subscribe(&self, user_id: i32, sub: PushSubscription) {
47 self.subscriptions
48 .write()
49 .await
50 .entry(user_id)
51 .or_default()
52 .push(sub);
53 tracing::info!(user_id, "push subscription added");
54 }
55
56 #[instrument(skip(self))]
57 pub async fn unsubscribe(&self, user_id: i32, endpoint: &str) -> bool {
58 self.subscriptions
59 .write()
60 .await
61 .get_mut(&user_id)
62 .is_some_and(|subs| {
63 let before = subs.len();
64 subs.retain(|s| s.endpoint != endpoint);
65 let removed = subs.len() < before;
66 if removed {
67 tracing::info!(user_id, "push subscription removed");
68 }
69 removed
70 })
71 }
72
73 #[instrument(skip(self))]
74 pub async fn notify_user(&self, user_id: i32, title: &str, body: &str) {
75 let Some(ref key) = self.vapid_private_key else {
76 tracing::warn!("VAPID not configured, cannot send push");
77 return;
78 };
79
80 let subs = self.subscriptions.read().await.get(&user_id).cloned();
81 let Some(subs) = subs else {
82 tracing::debug!(user_id, "no push subscriptions for user");
83 return;
84 };
85
86 let payload = serde_json::json!({
87 "title": title,
88 "body": body,
89 });
90 let payload_bytes = serde_json::to_vec(&payload).unwrap();
91
92 for sub in &subs {
93 let info = SubscriptionInfo::new(&sub.endpoint, &sub.p256dh, &sub.auth);
94 let sig = match VapidSignatureBuilder::from_base64(key, &info) {
95 Ok(mut b) => {
96 if let Some(ref contact) = self.vapid_contact {
97 b.add_claim("sub", contact.as_str());
98 }
99 match b.build() {
100 Ok(s) => s,
101 Err(e) => {
102 tracing::warn!(error = %e, "failed to build VAPID signature");
103 continue;
104 }
105 }
106 }
107 Err(e) => {
108 tracing::warn!(error = %e, "failed to create VAPID signature builder");
109 continue;
110 }
111 };
112
113 let mut msg = WebPushMessageBuilder::new(&info);
114 msg.set_payload(ContentEncoding::Aes128Gcm, &payload_bytes);
115 msg.set_vapid_signature(sig);
116
117 let client = IsahcWebPushClient::new().unwrap();
118 match client.send(msg.build().unwrap()).await {
119 Ok(()) => tracing::debug!("push sent to {}", sub.endpoint),
120 Err(WebPushError::EndpointNotValid(_)) => {
121 tracing::warn!("subscription expired, removing");
122 self.subscriptions
123 .write()
124 .await
125 .entry(user_id)
126 .or_default()
127 .retain(|s| s.endpoint != sub.endpoint);
128 }
129 Err(WebPushError::ServerError { retry_after, .. }) => {
130 tracing::warn!("push service overloaded, retry after {:?}", retry_after);
131 }
132 Err(e) => {
133 tracing::warn!(error = %e, "push send failed");
134 }
135 }
136 }
137 }
138}
139
140pub fn start_background_worker(task_state: AuthState) {
141 tokio::spawn(async move {
142 let mut interval = tokio::time::interval(std::time::Duration::from_hours(1));
143 loop {
144 interval.tick().await;
145 tracing::debug!("background worker: checking stale inbox tasks");
146
147 let tasks = if let Some(ref db) = task_state.db {
148 match db.list(0).await {
149 Ok(t) => t,
150 Err(_) => continue,
151 }
152 } else {
153 let repo = crate::repository::InMemoryTaskRepository::shared();
154 match repo.list(0).await {
155 Ok(t) => t,
156 Err(_) => continue,
157 }
158 };
159
160 let now = chrono::Utc::now();
161 let stale: Vec<_> = tasks
162 .iter()
163 .filter(|t| t.status == TaskStatus::Inbox && (now - t.created_at).num_hours() > 24)
164 .collect();
165
166 if stale.is_empty() {
167 continue;
168 }
169
170 tracing::info!(count = stale.len(), "found stale inbox tasks");
171 let msg = if stale.len() == 1 {
172 format!(
173 "Tienes 1 nota en el Inbox desde hace más de 24h: \"{}\"",
174 stale[0].description
175 )
176 } else {
177 format!(
178 "Tienes {} notas en el Inbox desde hace más de 24h. ¿Las clasificamos?",
179 stale.len()
180 )
181 };
182
183 let push = PushService::new();
184 push.notify_user(0, "oxinbox — Inbox estancado", &msg).await;
185 }
186 });
187}