sova_notifications/
notify.rs

1//! Send notifications to users / audiences.
2
3use crate::channel::{Channel, Via};
4use crate::entity;
5use crate::list::NotificationRow;
6use chrono::Utc;
7use sova_core::{Error, EventBus, Request, Result};
8use sova_db::{DbError, DbExt, DbHandle};
9use sea_orm::{ActiveModelTrait, Set};
10use serde_json::Value;
11use std::collections::HashMap;
12use std::sync::Arc;
13
14/// Recipient marker on the request (or derived from CurrentUser when `auth` feature).
15#[derive(Clone, Copy, Debug)]
16pub struct NotificationUser(pub i64);
17
18/// App-state service installed by [`super::Notifications`].
19#[derive(Clone)]
20pub struct NotificationService {
21    pub(crate) channels: Arc<HashMap<String, Channel>>,
22}
23
24impl NotificationService {
25    pub fn channel(&self, slug: &str) -> Option<&Channel> {
26        self.channels.get(slug)
27    }
28
29    pub fn ensure_channel(&self, slug: &str) -> Result<&Channel> {
30        self.channels.get(slug).ok_or_else(|| {
31            Error::BadRequest(format!("unknown notification channel `{slug}`"))
32        })
33    }
34}
35
36/// Fluent notification builder.
37pub struct Notify {
38    recipients: Recipients,
39    channel: String,
40    event: String,
41    title: String,
42    body: Option<String>,
43    data: Value,
44    vias: Vec<Via>,
45    /// When true, skip publish-permission checks (trusted server code).
46    system: bool,
47}
48
49enum Recipients {
50    Users(Vec<i64>),
51    #[cfg(feature = "auth")]
52    Role(String),
53    #[cfg(feature = "auth")]
54    Permission(String),
55}
56
57impl Notify {
58    pub fn to(user_id: i64) -> Self {
59        Self::to_many([user_id])
60    }
61
62    pub fn to_many(ids: impl IntoIterator<Item = i64>) -> Self {
63        Self {
64            recipients: Recipients::Users(ids.into_iter().collect()),
65            channel: "default".into(),
66            event: "notification".into(),
67            title: String::new(),
68            body: None,
69            data: Value::Object(Default::default()),
70            vias: default_vias(),
71            system: true,
72        }
73    }
74
75    #[cfg(feature = "auth")]
76    pub fn to_role(slug: impl Into<String>) -> Self {
77        Self {
78            recipients: Recipients::Role(slug.into()),
79            channel: "default".into(),
80            event: "notification".into(),
81            title: String::new(),
82            body: None,
83            data: Value::Object(Default::default()),
84            vias: default_vias(),
85            system: true,
86        }
87    }
88
89    #[cfg(feature = "auth")]
90    pub fn to_permission(slug: impl Into<String>) -> Self {
91        Self {
92            recipients: Recipients::Permission(slug.into()),
93            channel: "default".into(),
94            event: "notification".into(),
95            title: String::new(),
96            body: None,
97            data: Value::Object(Default::default()),
98            vias: default_vias(),
99            system: true,
100        }
101    }
102
103    pub fn channel(mut self, slug: impl Into<String>) -> Self {
104        self.channel = slug.into();
105        self
106    }
107
108    pub fn event(mut self, event: impl Into<String>) -> Self {
109        self.event = event.into();
110        self
111    }
112
113    pub fn title(mut self, title: impl Into<String>) -> Self {
114        self.title = title.into();
115        self
116    }
117
118    pub fn body(mut self, body: impl Into<String>) -> Self {
119        self.body = Some(body.into());
120        self
121    }
122
123    pub fn data(mut self, data: Value) -> Self {
124        self.data = data;
125        self
126    }
127
128    pub fn via(mut self, vias: impl IntoIterator<Item = Via>) -> Self {
129        self.vias = vias.into_iter().collect();
130        self
131    }
132
133    /// Mark as system send (default for builders) — skips channel publish ACL.
134    pub fn system(mut self, yes: bool) -> Self {
135        self.system = yes;
136        self
137    }
138
139    /// HTTP / untrusted send: enforce channel publish permission when set.
140    pub fn as_user(mut self) -> Self {
141        self.system = false;
142        self
143    }
144
145    pub async fn send(self, req: &Request) -> Result<Vec<NotificationRow>> {
146        let svc = req
147            .try_state::<NotificationService>()
148            .ok_or_else(|| Error::Internal("Notifications plugin not installed".into()))?;
149        let ch = svc.ensure_channel(&self.channel)?;
150
151        if !self.system {
152            if let Some(ref perm) = ch.publish {
153                #[cfg(feature = "auth")]
154                {
155                    use sova_auth::AuthExt;
156                    req.require_permission(perm)?;
157                }
158                #[cfg(not(feature = "auth"))]
159                {
160                    let _ = perm;
161                    return Err(Error::custom(
162                        403,
163                        "channel publish ACL requires notifications-auth",
164                    ));
165                }
166            }
167        }
168
169        let user_ids = resolve_recipients(req.db(), &self.recipients).await?;
170        let channel = self.channel.clone();
171        let event = self.event.clone();
172        let mut out = Vec::with_capacity(user_ids.len());
173        for uid in user_ids.iter().copied() {
174            match deliver_one(req, uid, &self).await {
175                Ok(Some(row)) => out.push(row),
176                Ok(None) => {}
177                Err(e) => tracing::warn!(error = %e, user_id = uid, "notification deliver failed"),
178            }
179        }
180        if let Some(bus) = req.try_state::<EventBus>() {
181            bus.dispatch(crate::NotificationSent {
182                channel,
183                event,
184                recipients: user_ids,
185            });
186        }
187        Ok(out)
188    }
189}
190
191fn default_vias() -> Vec<Via> {
192    #[cfg(feature = "ws")]
193    {
194        vec![Via::Database, Via::Ws]
195    }
196    #[cfg(not(feature = "ws"))]
197    {
198        vec![Via::Database]
199    }
200}
201
202async fn resolve_recipients(db: &DbHandle, recipients: &Recipients) -> Result<Vec<i64>> {
203    match recipients {
204        Recipients::Users(ids) => {
205            let _ = db;
206            Ok(ids.clone())
207        }
208        #[cfg(feature = "auth")]
209        Recipients::Role(slug) => crate::audience::user_ids_with_role(db, slug).await,
210        #[cfg(feature = "auth")]
211        Recipients::Permission(slug) => crate::audience::user_ids_with_permission(db, slug).await,
212    }
213}
214
215async fn deliver_one(req: &Request, user_id: i64, n: &Notify) -> Result<Option<NotificationRow>> {
216    let mut row = None;
217    for via in &n.vias {
218        match via {
219            Via::Database => {
220                row = Some(insert_row(req.db(), user_id, n).await?);
221            }
222            #[cfg(feature = "ws")]
223            Via::Ws => {
224                push_ws(req, user_id, row.as_ref(), n).await;
225            }
226            #[cfg(feature = "mail")]
227            Via::Mail => {
228                send_mail(req, user_id, n).await;
229            }
230        }
231    }
232    Ok(row)
233}
234
235async fn insert_row(db: &DbHandle, user_id: i64, n: &Notify) -> Result<NotificationRow> {
236    let props = serde_json::to_string(&n.data).unwrap_or_else(|_| "{}".into());
237    let model = entity::ActiveModel {
238        user_id: Set(user_id),
239        channel: Set(n.channel.clone()),
240        event: Set(n.event.clone()),
241        title: Set(n.title.clone()),
242        body: Set(n.body.clone()),
243        data: Set(props),
244        read_at: Set(None),
245        created_at: Set(Utc::now()),
246        ..Default::default()
247    };
248    let inserted = model
249        .insert(db)
250        .await
251        .map_err(|e| Error::from(DbError::from(e)))?;
252    Ok(NotificationRow::from(inserted))
253}
254
255#[cfg(feature = "ws")]
256async fn push_ws(req: &Request, user_id: i64, row: Option<&NotificationRow>, n: &Notify) {
257    let Some(ws) = req.try_state::<sova_ws::WsShared>() else {
258        return;
259    };
260    let payload = if let Some(r) = row {
261        serde_json::to_string(r).unwrap_or_default()
262    } else {
263        serde_json::json!({
264            "user_id": user_id,
265            "channel": n.channel,
266            "event": n.event,
267            "title": n.title,
268            "body": n.body,
269            "data": n.data,
270        })
271        .to_string()
272    };
273    ws.hub
274        .broadcast(
275            &format!("user:{user_id}"),
276            sova_ws::Message::Text(payload.into()),
277        )
278        .await;
279}
280
281#[cfg(feature = "mail")]
282async fn send_mail(req: &Request, _user_id: i64, n: &Notify) {
283    use sova_mail::{Email, MailClient};
284    let Some(client) = req.try_state::<MailClient>() else {
285        tracing::warn!("Via::Mail requested but Mail plugin missing");
286        return;
287    };
288    let Some(to) = n.data.get("email").and_then(|v| v.as_str()) else {
289        tracing::warn!("Via::Mail skipped: data.email missing");
290        return;
291    };
292    let text = n.body.clone().unwrap_or_else(|| n.title.clone());
293    let html = format!("<p>{}</p>", html_escape(&text));
294    let _ = client
295        .send(
296            Email::new()
297                .to(to)
298                .subject(&n.title)
299                .text(text)
300                .html(html),
301        )
302        .await;
303}
304
305#[cfg(feature = "mail")]
306fn html_escape(s: &str) -> String {
307    s.replace('&', "&amp;")
308        .replace('<', "&lt;")
309        .replace('>', "&gt;")
310}
311
312/// Request helpers.
313pub trait NotifyExt {
314    fn notifications(&self) -> Option<Arc<NotificationService>>;
315}
316
317impl NotifyExt for Request {
318    fn notifications(&self) -> Option<Arc<NotificationService>> {
319        self.try_state::<NotificationService>()
320    }
321}
sova_notifications/notify.rs

sova_notifications/
notify.rs
