1use super::*;
2use rpc::Notification;
3use sea_orm::{SelectColumns, TryInsertResult};
4use time::OffsetDateTime;
5use util::ResultExt;
6
7impl Database {
8 /// Inserts a record representing a user joining the chat for a given channel.
9 pub async fn join_channel_chat(
10 &self,
11 channel_id: ChannelId,
12 connection_id: ConnectionId,
13 user_id: UserId,
14 ) -> Result<()> {
15 self.transaction(|tx| async move {
16 let channel = self.get_channel_internal(channel_id, &tx).await?;
17 self.check_user_is_channel_participant(&channel, user_id, &tx)
18 .await?;
19 channel_chat_participant::ActiveModel {
20 id: ActiveValue::NotSet,
21 channel_id: ActiveValue::Set(channel_id),
22 user_id: ActiveValue::Set(user_id),
23 connection_id: ActiveValue::Set(connection_id.id as i32),
24 connection_server_id: ActiveValue::Set(ServerId(connection_id.owner_id as i32)),
25 }
26 .insert(&*tx)
27 .await?;
28 Ok(())
29 })
30 .await
31 }
32
33 /// Removes `channel_chat_participant` records associated with the given connection ID.
34 pub async fn channel_chat_connection_lost(
35 &self,
36 connection_id: ConnectionId,
37 tx: &DatabaseTransaction,
38 ) -> Result<()> {
39 channel_chat_participant::Entity::delete_many()
40 .filter(
41 Condition::all()
42 .add(
43 channel_chat_participant::Column::ConnectionServerId
44 .eq(connection_id.owner_id),
45 )
46 .add(channel_chat_participant::Column::ConnectionId.eq(connection_id.id)),
47 )
48 .exec(tx)
49 .await?;
50 Ok(())
51 }
52
53 /// Removes `channel_chat_participant` records associated with the given user ID so they
54 /// will no longer get chat notifications.
55 pub async fn leave_channel_chat(
56 &self,
57 channel_id: ChannelId,
58 connection_id: ConnectionId,
59 _user_id: UserId,
60 ) -> Result<()> {
61 self.transaction(|tx| async move {
62 channel_chat_participant::Entity::delete_many()
63 .filter(
64 Condition::all()
65 .add(
66 channel_chat_participant::Column::ConnectionServerId
67 .eq(connection_id.owner_id),
68 )
69 .add(channel_chat_participant::Column::ConnectionId.eq(connection_id.id))
70 .add(channel_chat_participant::Column::ChannelId.eq(channel_id)),
71 )
72 .exec(&*tx)
73 .await?;
74
75 Ok(())
76 })
77 .await
78 }
79
80 /// Retrieves the messages in the specified channel.
81 ///
82 /// Use `before_message_id` to paginate through the channel's messages.
83 pub async fn get_channel_messages(
84 &self,
85 channel_id: ChannelId,
86 user_id: UserId,
87 count: usize,
88 before_message_id: Option<MessageId>,
89 ) -> Result<Vec<proto::ChannelMessage>> {
90 self.transaction(|tx| async move {
91 let channel = self.get_channel_internal(channel_id, &tx).await?;
92 self.check_user_is_channel_participant(&channel, user_id, &tx)
93 .await?;
94
95 let mut condition =
96 Condition::all().add(channel_message::Column::ChannelId.eq(channel_id));
97
98 if let Some(before_message_id) = before_message_id {
99 condition = condition.add(channel_message::Column::Id.lt(before_message_id));
100 }
101
102 let rows = channel_message::Entity::find()
103 .filter(condition)
104 .order_by_desc(channel_message::Column::Id)
105 .limit(count as u64)
106 .all(&*tx)
107 .await?;
108
109 self.load_channel_messages(rows, &tx).await
110 })
111 .await
112 }
113
114 /// Returns the channel messages with the given IDs.
115 pub async fn get_channel_messages_by_id(
116 &self,
117 user_id: UserId,
118 message_ids: &[MessageId],
119 ) -> Result<Vec<proto::ChannelMessage>> {
120 self.transaction(|tx| async move {
121 let rows = channel_message::Entity::find()
122 .filter(channel_message::Column::Id.is_in(message_ids.iter().copied()))
123 .order_by_desc(channel_message::Column::Id)
124 .all(&*tx)
125 .await?;
126
127 let mut channels = HashMap::<ChannelId, channel::Model>::default();
128 for row in &rows {
129 channels.insert(
130 row.channel_id,
131 self.get_channel_internal(row.channel_id, &tx).await?,
132 );
133 }
134
135 for (_, channel) in channels {
136 self.check_user_is_channel_participant(&channel, user_id, &tx)
137 .await?;
138 }
139
140 let messages = self.load_channel_messages(rows, &tx).await?;
141 Ok(messages)
142 })
143 .await
144 }
145
146 async fn load_channel_messages(
147 &self,
148 rows: Vec<channel_message::Model>,
149 tx: &DatabaseTransaction,
150 ) -> Result<Vec<proto::ChannelMessage>> {
151 let mut messages = rows
152 .into_iter()
153 .map(|row| {
154 let nonce = row.nonce.as_u64_pair();
155 proto::ChannelMessage {
156 id: row.id.to_proto(),
157 sender_id: row.sender_id.to_proto(),
158 body: row.body,
159 timestamp: row.sent_at.assume_utc().unix_timestamp() as u64,
160 mentions: vec![],
161 nonce: Some(proto::Nonce {
162 upper_half: nonce.0,
163 lower_half: nonce.1,
164 }),
165 reply_to_message_id: row.reply_to_message_id.map(|id| id.to_proto()),
166 edited_at: row
167 .edited_at
168 .map(|t| t.assume_utc().unix_timestamp() as u64),
169 }
170 })
171 .collect::<Vec<_>>();
172 messages.reverse();
173
174 let mut mentions = channel_message_mention::Entity::find()
175 .filter(channel_message_mention::Column::MessageId.is_in(messages.iter().map(|m| m.id)))
176 .order_by_asc(channel_message_mention::Column::MessageId)
177 .order_by_asc(channel_message_mention::Column::StartOffset)
178 .stream(tx)
179 .await?;
180
181 let mut message_ix = 0;
182 while let Some(mention) = mentions.next().await {
183 let mention = mention?;
184 let message_id = mention.message_id.to_proto();
185 while let Some(message) = messages.get_mut(message_ix) {
186 if message.id < message_id {
187 message_ix += 1;
188 } else {
189 if message.id == message_id {
190 message.mentions.push(proto::ChatMention {
191 range: Some(proto::Range {
192 start: mention.start_offset as u64,
193 end: mention.end_offset as u64,
194 }),
195 user_id: mention.user_id.to_proto(),
196 });
197 }
198 break;
199 }
200 }
201 }
202
203 Ok(messages)
204 }
205
206 fn format_mentions_to_entities(
207 &self,
208 message_id: MessageId,
209 body: &str,
210 mentions: &[proto::ChatMention],
211 ) -> Result<Vec<tables::channel_message_mention::ActiveModel>> {
212 Ok(mentions
213 .iter()
214 .filter_map(|mention| {
215 let range = mention.range.as_ref()?;
216 if !body.is_char_boundary(range.start as usize)
217 || !body.is_char_boundary(range.end as usize)
218 {
219 return None;
220 }
221 Some(channel_message_mention::ActiveModel {
222 message_id: ActiveValue::Set(message_id),
223 start_offset: ActiveValue::Set(range.start as i32),
224 end_offset: ActiveValue::Set(range.end as i32),
225 user_id: ActiveValue::Set(UserId::from_proto(mention.user_id)),
226 })
227 })
228 .collect::<Vec<_>>())
229 }
230
231 /// Creates a new channel message.
232 #[allow(clippy::too_many_arguments)]
233 pub async fn create_channel_message(
234 &self,
235 channel_id: ChannelId,
236 user_id: UserId,
237 body: &str,
238 mentions: &[proto::ChatMention],
239 timestamp: OffsetDateTime,
240 nonce: u128,
241 reply_to_message_id: Option<MessageId>,
242 ) -> Result<CreatedChannelMessage> {
243 self.transaction(|tx| async move {
244 let channel = self.get_channel_internal(channel_id, &tx).await?;
245 self.check_user_is_channel_participant(&channel, user_id, &tx)
246 .await?;
247
248 let mut rows = channel_chat_participant::Entity::find()
249 .filter(channel_chat_participant::Column::ChannelId.eq(channel_id))
250 .stream(&*tx)
251 .await?;
252
253 let mut is_participant = false;
254 let mut participant_connection_ids = Vec::new();
255 let mut participant_user_ids = Vec::new();
256 while let Some(row) = rows.next().await {
257 let row = row?;
258 if row.user_id == user_id {
259 is_participant = true;
260 }
261 participant_user_ids.push(row.user_id);
262 participant_connection_ids.push(row.connection());
263 }
264 drop(rows);
265
266 if !is_participant {
267 Err(anyhow!("not a chat participant"))?;
268 }
269
270 let timestamp = timestamp.to_offset(time::UtcOffset::UTC);
271 let timestamp = time::PrimitiveDateTime::new(timestamp.date(), timestamp.time());
272
273 let result = channel_message::Entity::insert(channel_message::ActiveModel {
274 channel_id: ActiveValue::Set(channel_id),
275 sender_id: ActiveValue::Set(user_id),
276 body: ActiveValue::Set(body.to_string()),
277 sent_at: ActiveValue::Set(timestamp),
278 nonce: ActiveValue::Set(Uuid::from_u128(nonce)),
279 id: ActiveValue::NotSet,
280 reply_to_message_id: ActiveValue::Set(reply_to_message_id),
281 edited_at: ActiveValue::NotSet,
282 })
283 .on_conflict(
284 OnConflict::columns([
285 channel_message::Column::SenderId,
286 channel_message::Column::Nonce,
287 ])
288 .do_nothing()
289 .to_owned(),
290 )
291 .do_nothing()
292 .exec(&*tx)
293 .await?;
294
295 let message_id;
296 let mut notifications = Vec::new();
297 match result {
298 TryInsertResult::Inserted(result) => {
299 message_id = result.last_insert_id;
300 let mentioned_user_ids =
301 mentions.iter().map(|m| m.user_id).collect::<HashSet<_>>();
302
303 let mentions = self.format_mentions_to_entities(message_id, body, mentions)?;
304 if !mentions.is_empty() {
305 channel_message_mention::Entity::insert_many(mentions)
306 .exec(&*tx)
307 .await?;
308 }
309
310 for mentioned_user in mentioned_user_ids {
311 notifications.extend(
312 self.create_notification(
313 UserId::from_proto(mentioned_user),
314 rpc::Notification::ChannelMessageMention {
315 message_id: message_id.to_proto(),
316 sender_id: user_id.to_proto(),
317 channel_id: channel_id.to_proto(),
318 },
319 false,
320 &tx,
321 )
322 .await?,
323 );
324 }
325
326 self.observe_channel_message_internal(channel_id, user_id, message_id, &tx)
327 .await?;
328 }
329 _ => {
330 message_id = channel_message::Entity::find()
331 .filter(channel_message::Column::Nonce.eq(Uuid::from_u128(nonce)))
332 .one(&*tx)
333 .await?
334 .ok_or_else(|| anyhow!("failed to insert message"))?
335 .id;
336 }
337 }
338
339 let mut channel_members = self.get_channel_participants(&channel, &tx).await?;
340 channel_members.retain(|member| !participant_user_ids.contains(member));
341
342 Ok(CreatedChannelMessage {
343 message_id,
344 participant_connection_ids,
345 channel_members,
346 notifications,
347 })
348 })
349 .await
350 }
351
352 pub async fn observe_channel_message(
353 &self,
354 channel_id: ChannelId,
355 user_id: UserId,
356 message_id: MessageId,
357 ) -> Result<NotificationBatch> {
358 self.transaction(|tx| async move {
359 self.observe_channel_message_internal(channel_id, user_id, message_id, &tx)
360 .await?;
361 let mut batch = NotificationBatch::default();
362 batch.extend(
363 self.mark_notification_as_read(
364 user_id,
365 &Notification::ChannelMessageMention {
366 message_id: message_id.to_proto(),
367 sender_id: Default::default(),
368 channel_id: Default::default(),
369 },
370 &tx,
371 )
372 .await?,
373 );
374 Ok(batch)
375 })
376 .await
377 }
378
379 async fn observe_channel_message_internal(
380 &self,
381 channel_id: ChannelId,
382 user_id: UserId,
383 message_id: MessageId,
384 tx: &DatabaseTransaction,
385 ) -> Result<()> {
386 observed_channel_messages::Entity::insert(observed_channel_messages::ActiveModel {
387 user_id: ActiveValue::Set(user_id),
388 channel_id: ActiveValue::Set(channel_id),
389 channel_message_id: ActiveValue::Set(message_id),
390 })
391 .on_conflict(
392 OnConflict::columns([
393 observed_channel_messages::Column::ChannelId,
394 observed_channel_messages::Column::UserId,
395 ])
396 .update_column(observed_channel_messages::Column::ChannelMessageId)
397 .action_cond_where(observed_channel_messages::Column::ChannelMessageId.lt(message_id))
398 .to_owned(),
399 )
400 // TODO: Try to upgrade SeaORM so we don't have to do this hack around their bug
401 .exec_without_returning(tx)
402 .await?;
403 Ok(())
404 }
405
406 pub async fn observed_channel_messages(
407 &self,
408 channel_ids: &[ChannelId],
409 user_id: UserId,
410 tx: &DatabaseTransaction,
411 ) -> Result<Vec<proto::ChannelMessageId>> {
412 let rows = observed_channel_messages::Entity::find()
413 .filter(observed_channel_messages::Column::UserId.eq(user_id))
414 .filter(
415 observed_channel_messages::Column::ChannelId
416 .is_in(channel_ids.iter().map(|id| id.0)),
417 )
418 .all(tx)
419 .await?;
420
421 Ok(rows
422 .into_iter()
423 .map(|message| proto::ChannelMessageId {
424 channel_id: message.channel_id.to_proto(),
425 message_id: message.channel_message_id.to_proto(),
426 })
427 .collect())
428 }
429
430 pub async fn latest_channel_messages(
431 &self,
432 channel_ids: &[ChannelId],
433 tx: &DatabaseTransaction,
434 ) -> Result<Vec<proto::ChannelMessageId>> {
435 let mut values = String::new();
436 for id in channel_ids {
437 if !values.is_empty() {
438 values.push_str(", ");
439 }
440 write!(&mut values, "({})", id).unwrap();
441 }
442
443 if values.is_empty() {
444 return Ok(Vec::default());
445 }
446
447 let sql = format!(
448 r#"
449 SELECT
450 *
451 FROM (
452 SELECT
453 *,
454 row_number() OVER (
455 PARTITION BY channel_id
456 ORDER BY id DESC
457 ) as row_number
458 FROM channel_messages
459 WHERE
460 channel_id in ({values})
461 ) AS messages
462 WHERE
463 row_number = 1
464 "#,
465 );
466
467 let stmt = Statement::from_string(self.pool.get_database_backend(), sql);
468 let mut last_messages = channel_message::Model::find_by_statement(stmt)
469 .stream(tx)
470 .await?;
471
472 let mut results = Vec::new();
473 while let Some(result) = last_messages.next().await {
474 let message = result?;
475 results.push(proto::ChannelMessageId {
476 channel_id: message.channel_id.to_proto(),
477 message_id: message.id.to_proto(),
478 });
479 }
480
481 Ok(results)
482 }
483
484 fn get_notification_kind_id_by_name(&self, notification_kind: &str) -> Option<i32> {
485 self.notification_kinds_by_id
486 .iter()
487 .find(|(_, kind)| **kind == notification_kind)
488 .map(|kind| kind.0 .0)
489 }
490
491 /// Removes the channel message with the given ID.
492 pub async fn remove_channel_message(
493 &self,
494 channel_id: ChannelId,
495 message_id: MessageId,
496 user_id: UserId,
497 ) -> Result<(Vec<ConnectionId>, Vec<NotificationId>)> {
498 self.transaction(|tx| async move {
499 let mut rows = channel_chat_participant::Entity::find()
500 .filter(channel_chat_participant::Column::ChannelId.eq(channel_id))
501 .stream(&*tx)
502 .await?;
503
504 let mut is_participant = false;
505 let mut participant_connection_ids = Vec::new();
506 while let Some(row) = rows.next().await {
507 let row = row?;
508 if row.user_id == user_id {
509 is_participant = true;
510 }
511 participant_connection_ids.push(row.connection());
512 }
513 drop(rows);
514
515 if !is_participant {
516 Err(anyhow!("not a chat participant"))?;
517 }
518
519 let result = channel_message::Entity::delete_by_id(message_id)
520 .filter(channel_message::Column::SenderId.eq(user_id))
521 .exec(&*tx)
522 .await?;
523
524 if result.rows_affected == 0 {
525 let channel = self.get_channel_internal(channel_id, &tx).await?;
526 if self
527 .check_user_is_channel_admin(&channel, user_id, &tx)
528 .await
529 .is_ok()
530 {
531 let result = channel_message::Entity::delete_by_id(message_id)
532 .exec(&*tx)
533 .await?;
534 if result.rows_affected == 0 {
535 Err(anyhow!("no such message"))?;
536 }
537 } else {
538 Err(anyhow!("operation could not be completed"))?;
539 }
540 }
541
542 let notification_kind_id =
543 self.get_notification_kind_id_by_name("ChannelMessageMention");
544
545 let existing_notifications = notification::Entity::find()
546 .filter(notification::Column::EntityId.eq(message_id))
547 .filter(notification::Column::Kind.eq(notification_kind_id))
548 .select_column(notification::Column::Id)
549 .all(&*tx)
550 .await?;
551
552 let existing_notification_ids = existing_notifications
553 .into_iter()
554 .map(|notification| notification.id)
555 .collect();
556
557 // remove all the mention notifications for this message
558 notification::Entity::delete_many()
559 .filter(notification::Column::EntityId.eq(message_id))
560 .filter(notification::Column::Kind.eq(notification_kind_id))
561 .exec(&*tx)
562 .await?;
563
564 Ok((participant_connection_ids, existing_notification_ids))
565 })
566 .await
567 }
568
569 /// Updates the channel message with the given ID, body and timestamp(edited_at).
570 pub async fn update_channel_message(
571 &self,
572 channel_id: ChannelId,
573 message_id: MessageId,
574 user_id: UserId,
575 body: &str,
576 mentions: &[proto::ChatMention],
577 edited_at: OffsetDateTime,
578 ) -> Result<UpdatedChannelMessage> {
579 self.transaction(|tx| async move {
580 let channel = self.get_channel_internal(channel_id, &tx).await?;
581 self.check_user_is_channel_participant(&channel, user_id, &tx)
582 .await?;
583
584 let mut rows = channel_chat_participant::Entity::find()
585 .filter(channel_chat_participant::Column::ChannelId.eq(channel_id))
586 .stream(&*tx)
587 .await?;
588
589 let mut is_participant = false;
590 let mut participant_connection_ids = Vec::new();
591 let mut participant_user_ids = Vec::new();
592 while let Some(row) = rows.next().await {
593 let row = row?;
594 if row.user_id == user_id {
595 is_participant = true;
596 }
597 participant_user_ids.push(row.user_id);
598 participant_connection_ids.push(row.connection());
599 }
600 drop(rows);
601
602 if !is_participant {
603 Err(anyhow!("not a chat participant"))?;
604 }
605
606 let channel_message = channel_message::Entity::find_by_id(message_id)
607 .filter(channel_message::Column::SenderId.eq(user_id))
608 .one(&*tx)
609 .await?;
610
611 let Some(channel_message) = channel_message else {
612 Err(anyhow!("Channel message not found"))?
613 };
614
615 let edited_at = edited_at.to_offset(time::UtcOffset::UTC);
616 let edited_at = time::PrimitiveDateTime::new(edited_at.date(), edited_at.time());
617
618 let updated_message = channel_message::ActiveModel {
619 body: ActiveValue::Set(body.to_string()),
620 edited_at: ActiveValue::Set(Some(edited_at)),
621 reply_to_message_id: ActiveValue::Unchanged(channel_message.reply_to_message_id),
622 id: ActiveValue::Unchanged(message_id),
623 channel_id: ActiveValue::Unchanged(channel_id),
624 sender_id: ActiveValue::Unchanged(user_id),
625 sent_at: ActiveValue::Unchanged(channel_message.sent_at),
626 nonce: ActiveValue::Unchanged(channel_message.nonce),
627 };
628
629 let result = channel_message::Entity::update_many()
630 .set(updated_message)
631 .filter(channel_message::Column::Id.eq(message_id))
632 .filter(channel_message::Column::SenderId.eq(user_id))
633 .exec(&*tx)
634 .await?;
635 if result.rows_affected == 0 {
636 return Err(anyhow!(
637 "Attempted to edit a message (id: {message_id}) which does not exist anymore."
638 ))?;
639 }
640
641 // we have to fetch the old mentions,
642 // so we don't send a notification when the message has been edited that you are mentioned in
643 let old_mentions = channel_message_mention::Entity::find()
644 .filter(channel_message_mention::Column::MessageId.eq(message_id))
645 .all(&*tx)
646 .await?;
647
648 // remove all existing mentions
649 channel_message_mention::Entity::delete_many()
650 .filter(channel_message_mention::Column::MessageId.eq(message_id))
651 .exec(&*tx)
652 .await?;
653
654 let new_mentions = self.format_mentions_to_entities(message_id, body, mentions)?;
655 if !new_mentions.is_empty() {
656 // insert new mentions
657 channel_message_mention::Entity::insert_many(new_mentions)
658 .exec(&*tx)
659 .await?;
660 }
661
662 let mut update_mention_user_ids = HashSet::default();
663 let mut new_mention_user_ids =
664 mentions.iter().map(|m| m.user_id).collect::<HashSet<_>>();
665 // Filter out users that were mentioned before
666 for mention in &old_mentions {
667 if new_mention_user_ids.contains(&mention.user_id.to_proto()) {
668 update_mention_user_ids.insert(mention.user_id.to_proto());
669 }
670
671 new_mention_user_ids.remove(&mention.user_id.to_proto());
672 }
673
674 let notification_kind_id =
675 self.get_notification_kind_id_by_name("ChannelMessageMention");
676
677 let existing_notifications = notification::Entity::find()
678 .filter(notification::Column::EntityId.eq(message_id))
679 .filter(notification::Column::Kind.eq(notification_kind_id))
680 .all(&*tx)
681 .await?;
682
683 // determine which notifications should be updated or deleted
684 let mut deleted_notification_ids = HashSet::default();
685 let mut updated_mention_notifications = Vec::new();
686 for notification in existing_notifications {
687 if update_mention_user_ids.contains(¬ification.recipient_id.to_proto()) {
688 if let Some(notification) =
689 self::notifications::model_to_proto(self, notification).log_err()
690 {
691 updated_mention_notifications.push(notification);
692 }
693 } else {
694 deleted_notification_ids.insert(notification.id);
695 }
696 }
697
698 let mut notifications = Vec::new();
699 for mentioned_user in new_mention_user_ids {
700 notifications.extend(
701 self.create_notification(
702 UserId::from_proto(mentioned_user),
703 rpc::Notification::ChannelMessageMention {
704 message_id: message_id.to_proto(),
705 sender_id: user_id.to_proto(),
706 channel_id: channel_id.to_proto(),
707 },
708 false,
709 &tx,
710 )
711 .await?,
712 );
713 }
714
715 Ok(UpdatedChannelMessage {
716 message_id,
717 participant_connection_ids,
718 notifications,
719 reply_to_message_id: channel_message.reply_to_message_id,
720 timestamp: channel_message.sent_at,
721 deleted_mention_notification_ids: deleted_notification_ids
722 .into_iter()
723 .collect::<Vec<_>>(),
724 updated_mention_notifications,
725 })
726 })
727 .await
728 }
729}