1use std::collections::BTreeMap;
62use std::sync::Arc;
63
64use anyhow::anyhow;
65use serde::{Deserialize, Serialize};
66use serde_json::Value;
67
68use super::User;
69use crate::db::{DateTime, Db, now};
70use crate::i18n::with_locale;
71use crate::mail::Mail;
72use crate::queue::{Job, JobContext};
73use crate::toast::{ToastAction, ToastKind};
74use crate::{AppState, Result};
75
76#[derive(Debug, Clone, Copy, PartialEq, Eq)]
78#[non_exhaustive]
79pub enum Channel {
80 Mail,
82 Database,
85 Custom(&'static str),
88}
89
90#[derive(Debug, Clone, Default, Serialize, Deserialize)]
93#[non_exhaustive]
94pub struct Recipient {
95 user: Option<User>,
96 routes: BTreeMap<String, String>,
98 #[serde(default)]
100 language: Option<String>,
101}
102
103impl Recipient {
104 pub fn for_user(user: &User) -> Self {
106 Self {
107 user: Some(user.clone()),
108 routes: BTreeMap::new(),
109 language: None,
110 }
111 }
112
113 pub fn user(&self) -> Option<&User> {
115 self.user.as_ref()
116 }
117
118 pub fn to(channel: &str, address: impl Into<String>) -> Self {
120 Self::default().and(channel, address)
121 }
122
123 pub fn and(mut self, channel: &str, address: impl Into<String>) -> Self {
125 self.routes.insert(channel.to_owned(), address.into());
126 self
127 }
128
129 pub fn address(&self, channel: &str) -> Option<String> {
132 self.routes.get(channel).cloned().or_else(|| match channel {
133 "mail" => self.user.as_ref().map(|u| u.email.clone()),
134 _ => None,
135 })
136 }
137
138 pub(crate) fn has_address(&self, address: &str) -> bool {
140 self.routes.values().any(|a| a == address) || self.email().as_deref() == Some(address)
141 }
142
143 pub fn email(&self) -> Option<String> {
145 self.address("mail")
146 }
147
148 pub fn in_locale(mut self, locale: impl Into<String>) -> Self {
150 self.language = Some(locale.into());
151 self
152 }
153
154 pub fn locale(&self) -> Option<String> {
159 self.language.clone().or_else(|| {
160 self.user
161 .as_ref()?
162 .get::<String>("locale")
163 .filter(|l| !l.is_empty())
164 })
165 }
166}
167
168impl From<&User> for Recipient {
169 fn from(user: &User) -> Self {
170 Self::for_user(user)
171 }
172}
173
174impl From<&crate::AuthUser> for Recipient {
175 fn from(user: &crate::AuthUser) -> Self {
176 Self::for_user(user)
177 }
178}
179
180impl From<&Recipient> for Recipient {
181 fn from(recipient: &Recipient) -> Self {
182 recipient.clone()
183 }
184}
185
186pub trait Notification: Send + Sync {
188 fn kind(&self) -> &'static str;
190
191 fn channels(&self, to: &Recipient) -> Vec<Channel> {
195 let _ = to;
196 vec![Channel::Mail]
197 }
198
199 fn to_mail(&self, _to: &Recipient, _state: &AppState) -> Result<Mail> {
201 Err(anyhow!("notification `{}` has no mail version", self.kind()).into())
202 }
203
204 fn to_database(&self, to: &Recipient, state: &AppState) -> Result<Value> {
209 let _ = (to, state);
210 Ok(Value::Null)
211 }
212
213 fn to_channel(&self, channel: &str, to: &Recipient, state: &AppState) -> Result<Value> {
216 let _ = (to, state);
217 Err(anyhow!(
218 "notification `{}` has no version for the `{channel}` channel",
219 self.kind()
220 )
221 .into())
222 }
223}
224
225#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
231#[non_exhaustive]
232pub struct DatabaseMessage {
233 pub status: ToastKind,
235 pub title: String,
237 #[serde(default, skip_serializing_if = "Option::is_none")]
239 pub body: Option<String>,
240 #[serde(default, skip_serializing_if = "Option::is_none")]
242 pub url: Option<String>,
243 #[serde(default, skip_serializing_if = "Vec::is_empty")]
245 pub actions: Vec<ToastAction>,
246 #[serde(flatten)]
248 pub extra: serde_json::Map<String, Value>,
249}
250
251impl DatabaseMessage {
252 pub fn new(status: ToastKind, title: impl Into<String>) -> Self {
254 Self {
255 status,
256 title: title.into(),
257 body: None,
258 url: None,
259 actions: Vec::new(),
260 extra: serde_json::Map::new(),
261 }
262 }
263
264 pub fn success(title: impl Into<String>) -> Self {
266 Self::new(ToastKind::Success, title)
267 }
268
269 pub fn info(title: impl Into<String>) -> Self {
271 Self::new(ToastKind::Info, title)
272 }
273
274 pub fn warning(title: impl Into<String>) -> Self {
276 Self::new(ToastKind::Warning, title)
277 }
278
279 pub fn error(title: impl Into<String>) -> Self {
281 Self::new(ToastKind::Error, title)
282 }
283
284 pub fn body(mut self, body: impl Into<String>) -> Self {
286 self.body = Some(body.into());
287 self
288 }
289
290 pub fn url(mut self, url: impl Into<String>) -> Self {
292 self.url = Some(url.into());
293 self
294 }
295
296 pub fn action(mut self, action: ToastAction) -> Self {
298 self.actions.push(action);
299 self
300 }
301
302 pub fn link(self, label: impl Into<String>, url: impl Into<String>) -> Self {
304 self.action(ToastAction::link(label, url))
305 }
306
307 pub fn with(mut self, key: &str, value: impl Serialize) -> Self {
310 self.extra.insert(
311 key.to_owned(),
312 serde_json::to_value(value).unwrap_or(Value::Null),
313 );
314 self
315 }
316}
317
318impl From<DatabaseMessage> for Value {
319 fn from(message: DatabaseMessage) -> Self {
320 serde_json::to_value(message).unwrap_or(Value::Null)
321 }
322}
323
324#[derive(Debug, Clone, Serialize)]
326#[non_exhaustive]
327pub struct DatabaseNotification {
328 pub id: i64,
330 pub kind: String,
332 pub data: Value,
334 pub read_at: Option<DateTime>,
336 pub created_at: DateTime,
338}
339
340impl DatabaseNotification {
341 pub fn message(&self) -> Option<DatabaseMessage> {
344 serde_json::from_value(self.data.clone()).ok()
345 }
346}
347
348pub(crate) struct Hub {
354 tx: tokio::sync::broadcast::Sender<Signal>,
355}
356
357#[derive(Debug, Clone, PartialEq, Eq)]
358pub(crate) enum Signal {
359 User(i64),
361 Event(Option<i64>, Arc<Broadcast>),
363 Stop,
365}
366
367#[derive(Debug, PartialEq, Eq)]
370pub(crate) struct Broadcast {
371 pub(crate) event: String,
372 pub(crate) data: String,
373}
374
375impl Hub {
376 pub(crate) fn new() -> Self {
377 Self {
378 tx: tokio::sync::broadcast::channel(256).0,
379 }
380 }
381
382 pub(crate) fn touch(&self, user_id: i64) {
384 let _ = self.tx.send(Signal::User(user_id));
385 }
386
387 pub(crate) fn event(&self, user_id: Option<i64>, event: Broadcast) {
389 let _ = self.tx.send(Signal::Event(user_id, Arc::new(event)));
390 }
391
392 pub(crate) fn stop(&self) {
394 let _ = self.tx.send(Signal::Stop);
395 }
396
397 pub(crate) fn subscribe(&self) -> tokio::sync::broadcast::Receiver<Signal> {
398 self.tx.subscribe()
399 }
400}
401
402pub(crate) type ChannelFn = Arc<
403 dyn Fn(
404 AppState,
405 Recipient,
406 Value,
407 ) -> std::pin::Pin<Box<dyn std::future::Future<Output = Result> + Send>>
408 + Send
409 + Sync,
410>;
411
412pub(crate) fn channel_fn<F, Fut>(send: F) -> ChannelFn
413where
414 F: Fn(Recipient, Value, AppState) -> Fut + Send + Sync + 'static,
415 Fut: std::future::Future<Output = Result> + Send + 'static,
416{
417 Arc::new(move |state, to, message| Box::pin(send(to, message, state)))
418}
419
420#[derive(Serialize, Deserialize)]
422pub(crate) struct SendToChannel {
423 channel: String,
424 to: Recipient,
425 message: Value,
426}
427
428impl Job for SendToChannel {
429 const NAME: &'static str = "renox.send-to-channel";
430 const MAX_ATTEMPTS: u32 = 5;
431
432 async fn handle(self, ctx: JobContext) -> Result {
433 let send = ctx.state.channel(&self.channel)?;
434 send(ctx.state.clone(), self.to, self.message).await
435 }
436}
437
438fn ordered(notification: &impl Notification, to: &Recipient) -> Vec<Channel> {
442 let mut channels = notification.channels(to);
443 channels.sort_by_key(|channel| match channel {
444 Channel::Database => 0,
445 Channel::Custom(_) => 1,
446 _ => 2,
447 });
448 channels
449}
450
451fn event_name(name: &str) -> Result<&str> {
454 let valid = !name.is_empty()
455 && name.len() <= 100
456 && name
457 .chars()
458 .all(|c| c.is_ascii_alphanumeric() || matches!(c, '-' | '_' | ':' | '.'));
459 if valid {
460 Ok(name)
461 } else {
462 Err(anyhow!(
463 "`{name}` can't be a broadcast event's name: use letters, digits, `-`, `_`, `:` and `.`"
464 )
465 .into())
466 }
467}
468
469impl AppState {
470 pub fn broadcast(&self, event: &str, data: impl Serialize) -> Result {
495 self.send_broadcast(None, event, data)
496 }
497
498 pub fn broadcast_to(&self, user_id: i64, event: &str, data: impl Serialize) -> Result {
500 self.send_broadcast(Some(user_id), event, data)
501 }
502
503 fn send_broadcast(&self, user_id: Option<i64>, event: &str, data: impl Serialize) -> Result {
504 let event = event_name(event)?.to_owned();
505 let data = serde_json::to_value(&data).map_err(anyhow::Error::from)?;
506 let sent = crate::SentBroadcast {
507 user_id,
508 event,
509 data,
510 };
511 if self.fakes.record_broadcast(sent.clone()) {
512 return Ok(());
513 }
514 self.notification_hub.event(
515 user_id,
516 Broadcast {
517 event: sent.event,
518 data: sent.data.to_string(),
519 },
520 );
521 Ok(())
522 }
523
524 fn channel(&self, name: &str) -> Result<ChannelFn> {
525 self.channels.get(name).cloned().ok_or_else(|| {
526 anyhow!("no `{name}` notification channel: register it with `App::channel`").into()
527 })
528 }
529
530 async fn store_notification(&self, to: &Recipient, notification: &impl Notification) -> Result {
531 let Some(user) = &to.user else { return Ok(()) };
533 crate::db::sql(
534 "INSERT INTO notifications (user_id, kind, data, created_at) VALUES (?, ?, ?, ?)",
535 )
536 .bind(user.id)
537 .bind(notification.kind())
538 .bind(
539 with_locale(to.locale().as_deref(), || {
540 notification.to_database(to, self)
541 })?
542 .to_string(),
543 )
544 .bind(now())
545 .execute(&self.db)
546 .await?;
547 self.notification_hub.touch(user.id);
548 Ok(())
549 }
550
551 pub async fn notify(
556 &self,
557 to: impl Into<Recipient>,
558 notification: &impl Notification,
559 ) -> Result {
560 let to = &to.into();
561 if self.fakes.record_notification(notification.kind(), to) {
562 return Ok(());
563 }
564 let locale = to.locale();
565 let locale = locale.as_deref();
566 for channel in ordered(notification, to) {
567 match channel {
568 Channel::Database => self.store_notification(to, notification).await?,
569 Channel::Custom(name) => {
570 let send = self.channel(name)?;
571 let message = with_locale(locale, || notification.to_channel(name, to, self))?;
572 send(self.clone(), to.clone(), message).await?;
573 }
574 Channel::Mail => {
575 let mail = with_locale(locale, || notification.to_mail(to, self))?;
576 self.mailer.send(mail).await?;
577 }
578 }
579 }
580 Ok(())
581 }
582
583 pub async fn notify_later(
587 &self,
588 to: impl Into<Recipient>,
589 notification: &impl Notification,
590 ) -> Result {
591 let to = to.into();
592 if self.fakes.record_notification(notification.kind(), &to) {
593 return Ok(());
594 }
595 let locale = to.locale();
596 let locale = locale.as_deref();
597 for channel in ordered(notification, &to) {
598 match channel {
599 Channel::Database => self.store_notification(&to, notification).await?,
600 Channel::Custom(name) => {
601 self.channel(name)?; let message = with_locale(locale, || notification.to_channel(name, &to, self))?;
603 self.dispatch(SendToChannel {
604 channel: name.to_owned(),
605 to: to.clone(),
606 message,
607 })
608 .await?;
609 }
610 Channel::Mail => {
611 let mail = with_locale(locale, || notification.to_mail(&to, self))?;
612 self.queue_mail(mail).await?;
613 }
614 }
615 }
616 Ok(())
617 }
618}
619
620pub(super) fn from_row(row: &crate::db::Row) -> Result<DatabaseNotification> {
621 let data: String = row.try_get("data")?;
622 Ok(DatabaseNotification {
623 id: row.try_get("id")?,
624 kind: row.try_get("kind")?,
625 data: serde_json::from_str(&data).unwrap_or(Value::Null),
626 read_at: row.try_get("read_at")?,
627 created_at: row.try_get("created_at")?,
628 })
629}
630
631impl User {
632 pub async fn notifications(&self, db: &Db, limit: u32) -> Result<Vec<DatabaseNotification>> {
634 let rows = crate::db::sql(
635 "SELECT id, kind, data, read_at, created_at FROM notifications \
636 WHERE user_id = ? ORDER BY id DESC LIMIT ?",
637 )
638 .bind(self.id)
639 .bind(i64::from(limit))
640 .fetch_all(db)
641 .await?;
642 rows.iter().map(from_row).collect()
643 }
644
645 pub async fn notifications_before(
648 &self,
649 db: &Db,
650 before: i64,
651 limit: u32,
652 ) -> Result<Vec<DatabaseNotification>> {
653 let rows = crate::db::sql(
654 "SELECT id, kind, data, read_at, created_at FROM notifications \
655 WHERE user_id = ? AND id < ? ORDER BY id DESC LIMIT ?",
656 )
657 .bind(self.id)
658 .bind(before)
659 .bind(i64::from(limit))
660 .fetch_all(db)
661 .await?;
662 rows.iter().map(from_row).collect()
663 }
664
665 pub async fn notification(&self, db: &Db, id: i64) -> Result<Option<DatabaseNotification>> {
667 let row = crate::db::sql(
668 "SELECT id, kind, data, read_at, created_at FROM notifications \
669 WHERE id = ? AND user_id = ?",
670 )
671 .bind(id)
672 .bind(self.id)
673 .fetch_optional(db)
674 .await?;
675 row.as_ref().map(from_row).transpose()
676 }
677
678 pub async fn unread_notifications(&self, db: &Db) -> Result<Vec<DatabaseNotification>> {
680 let rows = crate::db::sql(
681 "SELECT id, kind, data, read_at, created_at FROM notifications \
682 WHERE user_id = ? AND read_at IS NULL ORDER BY id DESC",
683 )
684 .bind(self.id)
685 .fetch_all(db)
686 .await?;
687 rows.iter().map(from_row).collect()
688 }
689
690 pub async fn unread_notification_count(&self, db: &Db) -> Result<i64> {
692 Ok(crate::db::sql(
693 "SELECT COUNT(*) FROM notifications WHERE user_id = ? AND read_at IS NULL",
694 )
695 .bind(self.id)
696 .scalar(db)
697 .await?)
698 }
699
700 pub async fn mark_notification_read(&self, db: &Db, id: i64) -> Result<bool> {
702 let done = crate::db::sql(
703 "UPDATE notifications SET read_at = COALESCE(read_at, ?) WHERE id = ? AND user_id = ?",
704 )
705 .bind(now())
706 .bind(id)
707 .bind(self.id)
708 .execute(db)
709 .await?;
710 Ok(done > 0)
711 }
712
713 pub async fn mark_notification_unread(&self, db: &Db, id: i64) -> Result<bool> {
716 let done =
717 crate::db::sql("UPDATE notifications SET read_at = NULL WHERE id = ? AND user_id = ?")
718 .bind(id)
719 .bind(self.id)
720 .execute(db)
721 .await?;
722 Ok(done > 0)
723 }
724
725 pub async fn delete_notification(&self, db: &Db, id: i64) -> Result<bool> {
727 let done = crate::db::sql("DELETE FROM notifications WHERE id = ? AND user_id = ?")
728 .bind(id)
729 .bind(self.id)
730 .execute(db)
731 .await?;
732 Ok(done > 0)
733 }
734
735 pub async fn delete_notifications(&self, db: &Db) -> Result<u64> {
737 Ok(
738 crate::db::sql("DELETE FROM notifications WHERE user_id = ?")
739 .bind(self.id)
740 .execute(db)
741 .await?,
742 )
743 }
744
745 pub async fn mark_all_notifications_read(&self, db: &Db) -> Result<u64> {
747 let done = crate::db::sql(
748 "UPDATE notifications SET read_at = ? WHERE user_id = ? AND read_at IS NULL",
749 )
750 .bind(now())
751 .bind(self.id)
752 .execute(db)
753 .await?;
754 Ok(done)
755 }
756}
757
758pub async fn prune_read_notifications(db: &Db, age: std::time::Duration) -> Result<u64> {
763 let before = now() - chrono::Duration::from_std(age).unwrap_or_default();
764 Ok(
765 crate::db::sql("DELETE FROM notifications WHERE read_at < ?")
766 .bind(before)
767 .execute(db)
768 .await?,
769 )
770}