eeff292
use models::chat::room_seq;
use models::chat::room_user_settings::{self, RoomNotificationLevel};
use models::room::{room_members, rooms};
use models::users::user_notifications::NotificationType;
use models::users::user_recent_mentions;
use sea_orm::{
ActiveModelTrait, ColumnTrait, ConnectionTrait, DatabaseBackend, EntityTrait, QueryFilter, Set,
Statement,
};
use space::MessageNotifications;
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.database.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: &push::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 = push::notify::insert(db, recipient, ntype, title, body, data, actor_uid).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.database.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) = service.participants.access.member(space_id, uid).await?
{
if mute_active(member.muted, 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.
#[derive(Clone)]
pub struct EveryoneScope {
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: Vec<i64>,
pub request_id: Option<String>,
pub trace: Option<String>,
}
/// The recipients of an @everyone mention: room/space members minus the actor
/// and the explicitly mentioned users.
pub async fn everyone_candidates(service: &ChatService, scope: &EveryoneScope) -> Vec<i64> {
let room_id = scope.room_id;
let actor = scope.actor;
let candidates: Vec<i64> = match scope.space_id {
None => room_members::Entity::find()
.filter(room_members::Column::RoomId.eq(room_id))
.filter(room_members::Column::LeftAt.is_null())
.all(service.database.read())
.await
.unwrap_or_default()
.into_iter()
.map(|m| m.user_id)
.collect(),
Some(space_id) => {
let Ok(space) = service.participants.access.by_id(space_id).await else {
return Vec::new();
};
let mut ids: Vec<i64> = space
.members(actor)
.await
.unwrap_or_default()
.into_iter()
.filter(|m| !m.suppress_everyone)
.map(|m| m.user_id)
.collect();
if scope.is_private {
let mut allowed: Vec<i64> = models::room::room_members::Entity::find()
.filter(models::room::room_members::Column::RoomId.eq(room_id))
.filter(models::room::room_members::Column::LeftAt.is_null())
.all(service.database.read())
.await
.unwrap_or_default()
.into_iter()
.map(|row| row.user_id)
.collect();
if let Ok(Some(room)) = rooms::Entity::find_by_id(room_id)
.one(service.database.read())
.await
{
allowed.push(room.owner_id);
}
ids.retain(|uid| allowed.contains(uid));
}
ids
}
};
candidates
.into_iter()
.filter(|uid| *uid != actor && !scope.explicit.contains(uid))
.collect()
}
/// Fan out an @everyone mention to one bounded batch of recipients: writes
/// notifications and recent mentions, then pushes a WS user event.
pub async fn fanout_everyone_chunk(
service: &ChatService,
scope: &EveryoneScope,
recipients: &[i64],
) {
let data = serde_json::json!({
"room_id": scope.room_id,
"message_id": scope.message_id,
"everyone": true,
});
for &uid in recipients {
match notification_decision(service, scope.room_id, scope.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(scope.room_id),
message_id: Set(scope.message_id),
created_at: Set(now),
}
.insert(service.database.write())
.await;
let _ = service
.audience
.notifications
.create(
uid,
NotificationType::Mention,
i18n::catalog::notification_mention_message_title(
service.audience.recipient_locale(uid).await,
)
.to_string(),
Some(scope.body.clone()),
Some(data.clone()),
Some(scope.actor),
)
.await;
service.audience.realtime
.publish_user(
uid,
&serde_json::json!({"kind": "mention", "room_id": scope.room_id, "message_id": scope.message_id, "everyone": true}),
)
.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.database.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) = service.participants.access.member(space_id, uid).await?
{
if mute_active(member.muted, 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.audience.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())
}