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