use std::time::Duration;
use jiff::Timestamp;
use toasty::{Db, stmt::Expr};
use topcoat::context::{Cx, CxId, context};
use uuid::Uuid;
use super::Notification;
use crate::{
TenantId,
auth::current_realm,
bus::{NotificationBus, bus},
};
#[derive(Debug, Clone, toasty::Model)]
#[index(user_id, panel, created_at)]
pub struct StoredNotification {
#[key]
#[auto]
pub id: Uuid,
pub user_id: String,
pub panel: String,
pub tenant: Option<TenantId>,
#[column(type = text)]
pub notification: toasty::Json<Notification>,
pub read_at: Option<Timestamp>,
pub created_at: Timestamp,
}
#[derive(Clone)]
pub struct Notifications {
db: Db,
bus: NotificationBus,
panel: String,
tenant: Option<TenantId>,
request: Option<CxId>,
}
impl Notifications {
pub fn of(cx: &Cx) -> topcoat::Result<Self> {
let Some(realm) = current_realm(cx) else {
return Err(crate::error::declaration(
"Notifications::of needs a request under a panel".to_owned(),
));
};
let mut notifications = Self::new(
context::<Db>(cx).clone(),
bus(cx),
realm.prefix.clone(),
crate::tenant_id(cx),
);
notifications.request = Some(cx.id());
Ok(notifications)
}
pub(crate) fn new(
db: Db,
bus: NotificationBus,
panel: String,
tenant: Option<TenantId>,
) -> Self {
Self {
db,
bus,
panel,
tenant,
request: None,
}
}
#[must_use]
pub fn tenant(mut self, tenant: impl Into<Option<TenantId>>) -> Self {
self.tenant = tenant.into();
self
}
pub async fn send(
&self,
user_id: &str,
notification: Notification,
) -> topcoat::Result<StoredNotification> {
if let Some(request) = self.request {
crate::db::no_write_open(request, "Notifications::send")?;
}
let stored = self
.store(&mut self.db.clone(), user_id, notification)
.await?;
self.bus.wake(&self.panel, user_id);
Ok(stored)
}
pub async fn send_in(
&self,
ex: &mut dyn toasty::Executor,
user_id: &str,
notification: Notification,
) -> topcoat::Result<StoredNotification> {
let stored = self.store(ex, user_id, notification).await?;
let (bus, panel, user_id) = (self.bus.clone(), self.panel.clone(), user_id.to_owned());
let wake = move || bus.wake(&panel, &user_id);
let unqueued = match self.request {
Some(request) => crate::db::after_write_commits(request, wake),
None => Err(wake),
};
if let Err(wake) = unqueued {
wake();
}
Ok(stored)
}
async fn store(
&self,
ex: &mut dyn toasty::Executor,
user_id: &str,
notification: Notification,
) -> topcoat::Result<StoredNotification> {
toasty::create!(StoredNotification {
user_id: user_id.to_owned(),
panel: self.panel.clone(),
tenant: self.tenant,
notification: toasty::Json(notification),
read_at: None,
created_at: Timestamp::now(),
})
.exec(ex)
.await
.map_err(failure)
}
pub async fn prune(&self, age: Duration) -> topcoat::Result<()> {
let cutoff = Timestamp::now()
.checked_sub(jiff::SignedDuration::try_from(age).map_err(topcoat::Error::from)?)
.map_err(topcoat::Error::from)?;
let fields = StoredNotification::fields();
StoredNotification::filter(
fields
.panel()
.eq(self.panel.clone())
.and(fields.created_at().lt(cutoff)),
)
.delete()
.exec(&mut self.db.clone())
.await
.map_err(failure)
}
}
fn failure(error: toasty::Error) -> topcoat::Error {
crate::error::driver_failure(error, "Notifications are unavailable right now.")
}
#[derive(Clone)]
pub(crate) struct Inbox {
pub(crate) panel: String,
pub(crate) user_id: String,
tenant: Option<TenantId>,
}
impl Inbox {
pub(crate) fn of(cx: &Cx) -> Option<Self> {
let signed = crate::auth::signed(cx)?;
Some(Self {
panel: signed.realm.prefix.clone(),
user_id: signed.user.user_id(),
tenant: crate::tenant_id(cx),
})
}
fn holds(&self) -> Expr<bool> {
let fields = StoredNotification::fields();
let tenant = match self.tenant {
Some(tenant) => fields.tenant().is_none().or(fields.tenant().eq(tenant)),
None => fields.tenant().is_none(),
};
fields
.user_id()
.eq(self.user_id.clone())
.and(fields.panel().eq(self.panel.clone()))
.and(tenant)
}
fn holds_id(&self, id: Uuid) -> Expr<bool> {
StoredNotification::fields().id().eq(id).and(self.holds())
}
pub(crate) async fn latest(
&self,
db: &Db,
limit: usize,
) -> topcoat::Result<Vec<StoredNotification>> {
StoredNotification::filter(self.holds())
.order_by(StoredNotification::fields().created_at().desc())
.limit(limit)
.exec(&mut db.clone())
.await
.map_err(failure)
}
pub(crate) async fn unread(&self, db: &Db) -> topcoat::Result<u64> {
StoredNotification::filter(
self.holds()
.and(StoredNotification::fields().read_at().is_none()),
)
.count()
.exec(&mut db.clone())
.await
.map_err(failure)
}
pub(crate) async fn mark(&self, db: &Db, id: Uuid, read: bool) -> topcoat::Result<()> {
StoredNotification::filter(self.holds_id(id))
.update()
.read_at(read.then(Timestamp::now))
.exec(&mut db.clone())
.await
.map_err(failure)
}
pub(crate) async fn mark_all_read(&self, db: &Db) -> topcoat::Result<()> {
StoredNotification::filter(
self.holds()
.and(StoredNotification::fields().read_at().is_none()),
)
.update()
.read_at(Some(Timestamp::now()))
.exec(&mut db.clone())
.await
.map_err(failure)
}
pub(crate) async fn remove(&self, db: &Db, id: Uuid) -> topcoat::Result<()> {
StoredNotification::filter(self.holds_id(id))
.delete()
.exec(&mut db.clone())
.await
.map_err(failure)
}
pub(crate) async fn clear(&self, db: &Db) -> topcoat::Result<()> {
StoredNotification::filter(self.holds())
.delete()
.exec(&mut db.clone())
.await
.map_err(failure)
}
}