b453d5d
use models::chat::message_reminders::{self, ReminderStatus};
use models::chat::messages;
use models::room::rooms;
use models::users::user_notifications::NotificationType;
use sea_orm::sea_query::Expr;
use sea_orm::{
ActiveModelTrait, ColumnTrait, EntityTrait, QueryFilter, QueryOrder, QuerySelect, Set,
};
use serde::{Deserialize, Serialize};
use time::OffsetDateTime;
use utoipa::ToSchema;
use uuid::Uuid;
use crate::errors::ChatError;
use crate::services::service::ChatService;
use crate::share::lookup::{require_room_readable, space_slug_for};
use crate::share::notify::notify_user;
#[derive(Debug, Clone, Serialize, ToSchema)]
pub struct ReminderDto {
pub id: Uuid,
pub room_id: Uuid,
pub message_id: Uuid,
pub message_preview: Option<String>,
pub remind_at: OffsetDateTime,
pub note: Option<String>,
pub status: ReminderStatus,
pub created_at: OffsetDateTime,
}
#[derive(Debug, Clone, Deserialize, ToSchema)]
pub struct CreateReminderCmd {
#[serde(deserialize_with = "export::serde_datetime::deserialize_datetime")]
pub remind_at: OffsetDateTime,
#[serde(default)]
pub note: Option<String>,
}
#[derive(Debug, Clone, Deserialize, ToSchema)]
pub struct UpdateReminderCmd {
#[serde(
default,
deserialize_with = "export::serde_datetime::deserialize_opt_datetime"
)]
pub remind_at: Option<OffsetDateTime>,
#[serde(default, deserialize_with = "export::serde_patch::deserialize_opt_opt")]
pub note: Option<Option<String>>,
}
impl From<message_reminders::Model> for ReminderDto {
fn from(r: message_reminders::Model) -> Self {
Self {
id: r.id,
room_id: r.room_id,
message_id: r.message_id,
remind_at: r.remind_at,
note: r.note,
message_preview: None,
status: r.status,
created_at: r.created_at,
}
}
}
impl ChatService {
pub async fn create_reminder(
&self,
caller: i64,
room_id: Uuid,
message_id: Uuid,
cmd: CreateReminderCmd,
) -> Result<ReminderDto, ChatError> {
let _ctx = require_room_readable(self, room_id, caller).await?;
let _message = messages::Entity::find_by_id(message_id)
.filter(messages::Column::RoomId.eq(room_id))
.one(self.db().read())
.await?
.filter(|m| m.deleted_at.is_none())
.ok_or(ChatError::NotFound)?;
if cmd.remind_at <= OffsetDateTime::now_utc() {
return Err(ChatError::BadRequest(
"remind_at must be in the future".to_string(),
));
}
let row = message_reminders::ActiveModel {
id: Set(Uuid::now_v7()),
user_id: Set(caller),
room_id: Set(room_id),
message_id: Set(message_id),
remind_at: Set(cmd.remind_at),
note: Set(cmd.note.clone().filter(|n| !n.trim().is_empty())),
status: Set(ReminderStatus::Pending),
created_at: Set(OffsetDateTime::now_utc()),
}
.insert(self.db().write())
.await?;
Ok(ReminderDto::from(row))
}
pub async fn list_reminders(&self, caller: i64) -> Result<Vec<ReminderDto>, ChatError> {
let rows = message_reminders::Entity::find()
.filter(message_reminders::Column::UserId.eq(caller))
.filter(message_reminders::Column::Status.eq(ReminderStatus::Pending))
.order_by_asc(message_reminders::Column::RemindAt)
.all(self.db().read())
.await?;
let mut items: Vec<ReminderDto> =
rows.iter().map(|r| ReminderDto::from(r.clone())).collect();
let msg_ids: Vec<Uuid> = rows.iter().map(|r| r.message_id).collect();
if !msg_ids.is_empty() {
use models::chat::messages;
use sea_orm::{ColumnTrait as _, EntityTrait as _, QueryFilter as _};
let msgs = messages::Entity::find()
.filter(messages::Column::Id.is_in(msg_ids))
.all(self.db().read())
.await?;
let previews: std::collections::HashMap<Uuid, String> = msgs
.into_iter()
.map(|m| {
let mut p: String = m.content.chars().take(120).collect();
if p.ends_with(char::is_whitespace) {
p = p.trim_end().to_string();
}
(m.id, p)
})
.collect();
for item in items.iter_mut() {
item.message_preview = previews.get(&item.message_id).cloned();
}
}
Ok(items)
}
pub async fn update_reminder(
&self,
caller: i64,
reminder_id: Uuid,
cmd: UpdateReminderCmd,
) -> Result<ReminderDto, ChatError> {
let row = message_reminders::Entity::find_by_id(reminder_id)
.filter(message_reminders::Column::UserId.eq(caller))
.one(self.db().write())
.await?
.ok_or(ChatError::NotFound)?;
if row.status != ReminderStatus::Pending {
return Err(ChatError::Conflict("reminder not pending".to_string()));
}
if let Some(remind_at) = cmd.remind_at
&& remind_at <= OffsetDateTime::now_utc()
{
return Err(ChatError::BadRequest(
"remind_at must be in the future".to_string(),
));
}
let mut am: message_reminders::ActiveModel = row.into();
if let Some(remind_at) = cmd.remind_at {
am.remind_at = Set(remind_at);
}
if let Some(note) = cmd.note {
am.note = Set(note.filter(|n| !n.trim().is_empty()));
}
let updated = am.update(self.db().write()).await?;
Ok(ReminderDto::from(updated))
}
pub async fn cancel_reminder(&self, caller: i64, reminder_id: Uuid) -> Result<(), ChatError> {
let row = message_reminders::Entity::find_by_id(reminder_id)
.filter(message_reminders::Column::UserId.eq(caller))
.one(self.db().write())
.await?
.ok_or(ChatError::NotFound)?;
if row.status != ReminderStatus::Pending {
return Err(ChatError::Conflict("reminder not pending".to_string()));
}
let mut am: message_reminders::ActiveModel = row.into();
am.status = Set(ReminderStatus::Cancelled);
am.update(self.db().write()).await?;
Ok(())
}
#[tracing::instrument(name = "chat.deliver_reminders", skip_all)]
pub async fn deliver_due_reminders(&self) -> Result<u64, ChatError> {
let now = OffsetDateTime::now_utc();
// Recover claims abandoned by a crashed instance.
let _ = message_reminders::Entity::update_many()
.col_expr(
message_reminders::Column::Status,
Expr::value(ReminderStatus::Pending),
)
.filter(message_reminders::Column::Status.eq(ReminderStatus::Delivering))
.filter(
message_reminders::Column::RemindAt
.lt(now - time::Duration::minutes(crate::worker::STUCK_CLAIM_MINUTES)),
)
.exec(self.db().write())
.await?;
let due = message_reminders::Entity::find()
.filter(message_reminders::Column::Status.eq(ReminderStatus::Pending))
.filter(message_reminders::Column::RemindAt.lte(now))
.limit(200)
.all(self.db().write())
.await?;
let count = due.len() as u64;
let pending = export::notify::PendingNotifications::default();
for reminder in due {
// Atomic claim so concurrent replicas notify at most once.
let claimed = message_reminders::Entity::update_many()
.col_expr(
message_reminders::Column::Status,
Expr::value(ReminderStatus::Delivering),
)
.filter(message_reminders::Column::Id.eq(reminder.id))
.filter(message_reminders::Column::Status.eq(ReminderStatus::Pending))
.exec(self.db().write())
.await?;
if claimed.rows_affected == 0 {
continue;
}
let space_id = rooms::Entity::find_by_id(reminder.room_id)
.one(self.db().read())
.await
.ok()
.flatten()
.and_then(|room| room.space_id);
let data = serde_json::json!({
"room_id": reminder.room_id,
"message_id": reminder.message_id,
"space_name": space_slug_for(self, space_id).await,
});
if let Err(error) = notify_user(
self.db().write(),
&pending,
reminder.user_id,
NotificationType::System,
i18n::catalog::notification_reminder_title(
self.recipient_locale(reminder.user_id).await,
)
.to_string(),
None,
Some(data),
None,
)
.await
{
tracing::warn!(%error, "failed to write reminder notification");
message_reminders::Entity::update_many()
.col_expr(
message_reminders::Column::Status,
Expr::value(ReminderStatus::Pending),
)
.filter(message_reminders::Column::Id.eq(reminder.id))
.filter(message_reminders::Column::Status.eq(ReminderStatus::Delivering))
.exec(self.db().write())
.await?;
continue;
}
message_reminders::Entity::update_many()
.col_expr(
message_reminders::Column::Status,
Expr::value(ReminderStatus::Done),
)
.filter(message_reminders::Column::Id.eq(reminder.id))
.filter(message_reminders::Column::Status.eq(ReminderStatus::Delivering))
.exec(self.db().write())
.await?;
self.realtime
.publish_user(
reminder.user_id,
&serde_json::json!({
"kind": "reminder",
"room_id": reminder.room_id,
"message_id": reminder.message_id,
}),
)
.await;
pending.flush(&self.state).await;
}
Ok(count)
}
}