ef0b83a
use models::chat::room_seq;
use models::chat::room_user_settings::{self, RoomNotificationLevel};
use models::room::room_members;
use models::space::MessageNotifications;
use models::users::user_notifications::{self, NotificationType};
use models::users::user_recent_mentions;
use sea_orm::{
ActiveModelTrait, ColumnTrait, ConnectionTrait, DatabaseBackend, EntityTrait, QueryFilter, Set,
Statement,
};
use time::OffsetDateTime;
use uuid::Uuid;
use crate::errors::ChatError;
use crate::services::service::ChatService;
use crate::share::realtime::RoomEvent;
/// The recipient set of a DM / group DM message: current room members minus the
/// author. Used for OS pushes, which (unlike mentions) do not write inbox rows.
pub async fn dm_recipients(service: &ChatService, room_id: Uuid, author: i64) -> Vec<i64> {
room_members::Entity::find()
.filter(room_members::Column::RoomId.eq(room_id))
.filter(room_members::Column::LeftAt.is_null())
.all(service.db().read())
.await
.unwrap_or_default()
.into_iter()
.map(|member| member.user_id)
.filter(|uid| *uid != author)
.collect()
}
pub async fn next_room_seq(db: &impl ConnectionTrait, room_id: Uuid) -> Result<i64, ChatError> {
let now = OffsetDateTime::now_utc();
let row = db
.query_one_raw(Statement::from_sql_and_values(
DatabaseBackend::Postgres,
r#"INSERT INTO chat_room_seq (room_id, seq, updated_at)
VALUES ($1, 1, $2)
ON CONFLICT (room_id)
DO UPDATE SET seq = chat_room_seq.seq + 1, updated_at = $2
RETURNING seq"#,
[room_id.into(), now.into()],
))
.await?;
let row = row.ok_or_else(|| ChatError::Internal("seq allocation returned no row".into()))?;
let seq: i64 = row.try_get("", "seq")?;
Ok(seq)
}
/// Current room tip without consuming a seq; used for non-message events so
/// edits/pins/deletes never inflate unread counters or reorder listings.
pub async fn current_room_seq(db: &impl ConnectionTrait, room_id: Uuid) -> i64 {
room_seq::Entity::find_by_id(room_id)
.one(db)
.await
.ok()
.flatten()
.map(|s| s.seq)
.unwrap_or(0)
}
pub async fn record_mentions(
db: &impl ConnectionTrait,
room_id: Uuid,
message_id: Uuid,
uids: &[i64],
) -> Result<(), ChatError> {
let now = OffsetDateTime::now_utc();
for &uid in uids {
user_recent_mentions::ActiveModel {
id: Set(Uuid::now_v7()),
user_id: Set(uid),
channel_id: Set(room_id),
message_id: Set(message_id),
created_at: Set(now),
}
.insert(db)
.await?;
}
Ok(())
}
#[allow(clippy::too_many_arguments)]
pub async fn notify_user(
db: &impl ConnectionTrait,
pending: &export::notify::PendingNotifications,
recipient: i64,
ntype: NotificationType,
title: String,
body: Option<String>,
data: Option<serde_json::Value>,
actor_uid: Option<i64>,
) -> Result<(), ChatError> {
// The row stores already-rendered text: there is no schema column for the
// language, so callers resolve it per recipient and pass finished copy in.
let row = user_notifications::ActiveModel {
id: Set(Uuid::now_v7()),
user_id: Set(recipient),
notification_type: Set(ntype),
title: Set(title),
body: Set(body),
data_json: Set(data),
actor_uid: Set(actor_uid),
read_at: Set(None),
created_at: Set(OffsetDateTime::now_utc()),
}
.insert(db)
.await?;
pending.push(recipient, row.to_event_value());
Ok(())
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum NotifyDecision {
Notify,
Skip,
}
pub(crate) fn mute_active(muted: bool, mute_until: Option<OffsetDateTime>) -> bool {
muted && mute_until.is_none_or(|until| until > OffsetDateTime::now_utc())
}
pub async fn notification_decision(
service: &ChatService,
room_id: Uuid,
space_id: Option<Uuid>,
uid: i64,
mentioned: bool,
everyone: bool,
) -> Result<NotifyDecision, ChatError> {
if let Some(setting) = room_user_settings::Entity::find_by_id((uid, room_id))
.one(service.db().read())
.await?
{
if mute_active(setting.muted, setting.mute_until) {
return Ok(if mentioned && !everyone {
NotifyDecision::Notify
} else {
NotifyDecision::Skip
});
}
match setting.notification_level {
RoomNotificationLevel::All => return Ok(NotifyDecision::Notify),
RoomNotificationLevel::Mentions | RoomNotificationLevel::Nothing => {
return Ok(if mentioned && !everyone {
NotifyDecision::Notify
} else {
NotifyDecision::Skip
});
}
RoomNotificationLevel::Default => {}
}
}
if let Some(space_id) = space_id
&& let Some(member) = space::authority::member(&service.state, uid, space_id, uid).await?
{
if mute_active(
member.muted,
space::authority::to_time_opt(member.mute_until),
) && !(mentioned && !everyone)
{
return Ok(NotifyDecision::Skip);
}
return Ok(match member.message_notifications {
MessageNotifications::All => NotifyDecision::Notify,
MessageNotifications::Mentions | MessageNotifications::Nothing => {
if mentioned && (!everyone || !member.suppress_everyone) {
NotifyDecision::Notify
} else {
NotifyDecision::Skip
}
}
});
}
Ok(NotifyDecision::Notify)
}
/// One @everyone fan-out. Grouped into a struct because the room scope, the
/// message and the actor are all `Uuid`-shaped and would be easy to transpose.
pub struct EveryoneMention<'a> {
pub room_id: Uuid,
pub space_id: Option<Uuid>,
pub is_private: bool,
pub message_id: Uuid,
pub body: String,
pub actor: i64,
pub explicit: &'a [i64],
}
/// Fan out an @everyone mention to room/space members that were not explicitly
/// mentioned. Runs best-effort after commit: writes notifications and recent
/// mentions, then pushes a WS user event.
pub async fn fanout_everyone_mentions(service: &ChatService, scope: EveryoneMention<'_>) {
let EveryoneMention {
room_id,
space_id,
is_private,
message_id,
body,
actor,
explicit,
} = scope;
let candidates: Vec<i64> = if is_private || space_id.is_none() {
room_members::Entity::find()
.filter(room_members::Column::RoomId.eq(room_id))
.filter(room_members::Column::LeftAt.is_null())
.all(service.db().read())
.await
.unwrap_or_default()
.into_iter()
.map(|m| m.user_id)
.collect()
} else if let Some(space_id) = space_id {
space::authority::list_members(&service.state, actor, space_id, false, false)
.await
.unwrap_or_default()
.into_iter()
.filter(|m| !m.suppress_everyone)
.map(|m| m.user_id)
.collect()
} else {
Vec::new()
};
let data = serde_json::json!({
"room_id": room_id,
"message_id": message_id,
"everyone": true,
});
let pending = export::notify::PendingNotifications::default();
for uid in candidates {
if uid == actor || explicit.contains(&uid) {
continue;
}
match notification_decision(service, room_id, space_id, uid, true, true).await {
Ok(NotifyDecision::Notify) => {}
_ => continue,
}
let now = OffsetDateTime::now_utc();
let _ = user_recent_mentions::ActiveModel {
id: Set(Uuid::now_v7()),
user_id: Set(uid),
channel_id: Set(room_id),
message_id: Set(message_id),
created_at: Set(now),
}
.insert(service.db().write())
.await;
let _ = notify_user(
service.db().write(),
&pending,
uid,
NotificationType::Mention,
i18n::catalog::notification_mention_message_title(service.recipient_locale(uid).await)
.to_string(),
Some(body.clone()),
Some(data.clone()),
Some(actor),
)
.await;
service
.realtime
.publish_user(
uid,
&serde_json::json!({"kind": "mention", "room_id": room_id, "message_id": message_id, "everyone": true}),
)
.await;
}
pending.flush(&service.state).await;
}
/// Mention e-mails honour per-room / per-space mute and notification level:
/// Nothing, muted (even mid-window) and blocked settings opt out.
pub async fn mention_email_allowed(
service: &ChatService,
room_id: Uuid,
space_id: Option<Uuid>,
uid: i64,
) -> Result<bool, ChatError> {
if let Some(setting) = room_user_settings::Entity::find_by_id((uid, room_id))
.one(service.db().read())
.await?
{
if mute_active(setting.muted, setting.mute_until) {
return Ok(false);
}
match setting.notification_level {
RoomNotificationLevel::Nothing => return Ok(false),
RoomNotificationLevel::All | RoomNotificationLevel::Mentions => return Ok(true),
RoomNotificationLevel::Default => {}
}
}
if let Some(space_id) = space_id
&& let Some(member) = space::authority::member(&service.state, uid, space_id, uid).await?
{
if mute_active(
member.muted,
space::authority::to_time_opt(member.mute_until),
) {
return Ok(false);
}
return Ok(!matches!(
member.message_notifications,
MessageNotifications::Nothing
));
}
Ok(true)
}
pub async fn publish_message_event(
service: &ChatService,
room_id: Uuid,
seq: i64,
message: serde_json::Value,
) {
let event = RoomEvent::MessagePublished {
room_id,
seq,
message,
};
if let Ok(payload) = serde_json::to_value(&event) {
crate::services::outbound::enqueue_outbound_event(
service,
room_id,
"message_published",
&payload,
)
.await;
}
service.realtime.publish_room(&event).await;
}
pub async fn advance_read_state(
db: &impl ConnectionTrait,
uid: i64,
room_id: Uuid,
message_id: Option<Uuid>,
seq: i64,
) -> Result<bool, ChatError> {
let now = OffsetDateTime::now_utc();
let row = db
.query_one_raw(Statement::from_sql_and_values(
DatabaseBackend::Postgres,
r#"INSERT INTO chat_read_state
(user_id, room_id, last_read_message_id, last_read_seq, last_viewed_at, updated_at)
VALUES ($1, $2, $3, $4, $5, $5)
ON CONFLICT (user_id, room_id) DO UPDATE SET
last_read_message_id = EXCLUDED.last_read_message_id,
last_read_seq = EXCLUDED.last_read_seq,
last_viewed_at = EXCLUDED.last_viewed_at,
updated_at = EXCLUDED.updated_at
WHERE chat_read_state.last_read_seq < EXCLUDED.last_read_seq
RETURNING room_id"#,
[uid.into(), room_id.into(), message_id.into(), seq.into(), now.into()],
))
.await?;
Ok(row.is_some())
}