26af08e
use std::sync::{Arc, Mutex};
use queue::Labels;
use serde_json::Value;
use crate::{AppDatabase, AppRedis};
#[derive(Clone)]
pub struct Notifications {
database: Arc<AppDatabase>,
redis: AppRedis,
}
impl Notifications {
pub fn new(database: Arc<AppDatabase>, redis: AppRedis) -> Self {
Self { database, redis }
}
}
impl Notifications {
#[allow(clippy::too_many_arguments)]
pub async fn create(
&self,
recipient: i64,
kind: models::users::user_notifications::NotificationType,
title: String,
body: Option<String>,
data: Option<Value>,
actor: Option<i64>,
) -> Result<models::users::user_notifications::Model, sea_orm::DbErr> {
use sea_orm::TransactionTrait;
let row = self
.database
.write()
.transaction::<_, models::users::user_notifications::Model, sea_orm::DbErr>(move |tx| {
Box::pin(async move { insert(tx, recipient, kind, title, body, data, actor).await })
})
.await
.map_err(|error| match error {
sea_orm::TransactionError::Connection(error)
| sea_orm::TransactionError::Transaction(error) => error,
})?;
self.deliver(recipient, row.to_event_value()).await;
Ok(row)
}
pub async fn deliver(&self, uid: i64, notification: Value) {
if let Some(job) = crate::push::PushJob::from_notification(uid, ¬ification) {
crate::push::enqueue(&self.redis, &job).await;
}
}
}
/// Notifications created inside a DB transaction; flush after commit.
#[derive(Clone, Default)]
pub struct PendingNotifications {
inner: Arc<Mutex<Vec<(i64, Value)>>>,
}
impl PendingNotifications {
pub fn push(&self, uid: i64, notification: Value) {
match self.inner.lock() {
Ok(mut pending) => pending.push((uid, notification)),
Err(_) => {
tracing::warn!("pending notifications lock poisoned");
}
}
}
pub async fn flush(&self, notifications: &Notifications) {
let batch: Vec<(i64, Value)> = match self.inner.lock() {
Ok(mut pending) => std::mem::take(&mut *pending),
Err(_) => return,
};
for (uid, event) in batch {
notifications.deliver(uid, event).await;
}
}
}
#[allow(clippy::too_many_arguments)]
pub async fn insert(
tx: &impl sea_orm::ConnectionTrait,
recipient: i64,
kind: models::users::user_notifications::NotificationType,
title: String,
body: Option<String>,
data: Option<Value>,
actor: Option<i64>,
) -> Result<models::users::user_notifications::Model, sea_orm::DbErr> {
use sea_orm::{ActiveModelTrait, Set};
let row = models::users::user_notifications::ActiveModel {
id: Set(uuid::Uuid::now_v7()),
user_id: Set(recipient),
notification_type: Set(kind),
title: Set(title),
body: Set(body),
data_json: Set(data),
actor_uid: Set(actor),
read_at: Set(None),
created_at: Set(time::OffsetDateTime::now_utc()),
}
.insert(tx)
.await?;
let labels = Labels::new()
.with(events::label::RECIPIENT_USER_ID, recipient)
.with(events::label::USER_ID, recipient);
events::emit(
tx,
"notify.notification.create",
labels,
serde_json::json!({ "kind": "notification", "notification": row.to_event_value() }),
)
.await?;
Ok(row)
}