1use super::*;
2use time::OffsetDateTime;
3
4impl Database {
5 pub async fn join_channel_chat(
6 &self,
7 channel_id: ChannelId,
8 connection_id: ConnectionId,
9 user_id: UserId,
10 ) -> Result<()> {
11 self.transaction(|tx| async move {
12 self.check_user_is_channel_member(channel_id, user_id, &*tx)
13 .await?;
14 channel_chat_participant::ActiveModel {
15 id: ActiveValue::NotSet,
16 channel_id: ActiveValue::Set(channel_id),
17 user_id: ActiveValue::Set(user_id),
18 connection_id: ActiveValue::Set(connection_id.id as i32),
19 connection_server_id: ActiveValue::Set(ServerId(connection_id.owner_id as i32)),
20 }
21 .insert(&*tx)
22 .await?;
23 Ok(())
24 })
25 .await
26 }
27
28 pub async fn channel_chat_connection_lost(
29 &self,
30 connection_id: ConnectionId,
31 tx: &DatabaseTransaction,
32 ) -> Result<()> {
33 channel_chat_participant::Entity::delete_many()
34 .filter(
35 Condition::all()
36 .add(
37 channel_chat_participant::Column::ConnectionServerId
38 .eq(connection_id.owner_id),
39 )
40 .add(channel_chat_participant::Column::ConnectionId.eq(connection_id.id)),
41 )
42 .exec(tx)
43 .await?;
44 Ok(())
45 }
46
47 pub async fn leave_channel_chat(
48 &self,
49 channel_id: ChannelId,
50 connection_id: ConnectionId,
51 _user_id: UserId,
52 ) -> Result<()> {
53 self.transaction(|tx| async move {
54 channel_chat_participant::Entity::delete_many()
55 .filter(
56 Condition::all()
57 .add(
58 channel_chat_participant::Column::ConnectionServerId
59 .eq(connection_id.owner_id),
60 )
61 .add(channel_chat_participant::Column::ConnectionId.eq(connection_id.id))
62 .add(channel_chat_participant::Column::ChannelId.eq(channel_id)),
63 )
64 .exec(&*tx)
65 .await?;
66
67 Ok(())
68 })
69 .await
70 }
71
72 pub async fn get_channel_messages(
73 &self,
74 channel_id: ChannelId,
75 user_id: UserId,
76 count: usize,
77 before_message_id: Option<MessageId>,
78 ) -> Result<Vec<proto::ChannelMessage>> {
79 self.transaction(|tx| async move {
80 self.check_user_is_channel_member(channel_id, user_id, &*tx)
81 .await?;
82
83 let mut condition =
84 Condition::all().add(channel_message::Column::ChannelId.eq(channel_id));
85
86 if let Some(before_message_id) = before_message_id {
87 condition = condition.add(channel_message::Column::Id.lt(before_message_id));
88 }
89
90 let mut rows = channel_message::Entity::find()
91 .filter(condition)
92 .order_by_desc(channel_message::Column::Id)
93 .limit(count as u64)
94 .stream(&*tx)
95 .await?;
96
97 let mut messages = Vec::new();
98 while let Some(row) = rows.next().await {
99 let row = row?;
100 let nonce = row.nonce.as_u64_pair();
101 messages.push(proto::ChannelMessage {
102 id: row.id.to_proto(),
103 sender_id: row.sender_id.to_proto(),
104 body: row.body,
105 timestamp: row.sent_at.assume_utc().unix_timestamp() as u64,
106 nonce: Some(proto::Nonce {
107 upper_half: nonce.0,
108 lower_half: nonce.1,
109 }),
110 });
111 }
112 drop(rows);
113 messages.reverse();
114 Ok(messages)
115 })
116 .await
117 }
118
119 pub async fn create_channel_message(
120 &self,
121 channel_id: ChannelId,
122 user_id: UserId,
123 body: &str,
124 timestamp: OffsetDateTime,
125 nonce: u128,
126 ) -> Result<(MessageId, Vec<ConnectionId>, Vec<UserId>)> {
127 self.transaction(|tx| async move {
128 let mut rows = channel_chat_participant::Entity::find()
129 .filter(channel_chat_participant::Column::ChannelId.eq(channel_id))
130 .stream(&*tx)
131 .await?;
132
133 let mut is_participant = false;
134 let mut participant_connection_ids = Vec::new();
135 let mut participant_user_ids = Vec::new();
136 while let Some(row) = rows.next().await {
137 let row = row?;
138 if row.user_id == user_id {
139 is_participant = true;
140 }
141 participant_user_ids.push(row.user_id);
142 participant_connection_ids.push(row.connection());
143 }
144 drop(rows);
145
146 if !is_participant {
147 Err(anyhow!("not a chat participant"))?;
148 }
149
150 let timestamp = timestamp.to_offset(time::UtcOffset::UTC);
151 let timestamp = time::PrimitiveDateTime::new(timestamp.date(), timestamp.time());
152
153 let message = channel_message::Entity::insert(channel_message::ActiveModel {
154 channel_id: ActiveValue::Set(channel_id),
155 sender_id: ActiveValue::Set(user_id),
156 body: ActiveValue::Set(body.to_string()),
157 sent_at: ActiveValue::Set(timestamp),
158 nonce: ActiveValue::Set(Uuid::from_u128(nonce)),
159 id: ActiveValue::NotSet,
160 })
161 .on_conflict(
162 OnConflict::column(channel_message::Column::Nonce)
163 .update_column(channel_message::Column::Nonce)
164 .to_owned(),
165 )
166 .exec(&*tx)
167 .await?;
168
169 #[derive(Debug, Clone, Copy, EnumIter, DeriveColumn)]
170 enum QueryConnectionId {
171 ConnectionId,
172 }
173
174 // Observe this message for the sender
175 self.observe_channel_message_internal(
176 channel_id,
177 user_id,
178 message.last_insert_id,
179 &*tx,
180 )
181 .await?;
182
183 let mut channel_members = self.get_channel_members_internal(channel_id, &*tx).await?;
184 channel_members.retain(|member| !participant_user_ids.contains(member));
185
186 Ok((
187 message.last_insert_id,
188 participant_connection_ids,
189 channel_members,
190 ))
191 })
192 .await
193 }
194
195 pub async fn observe_channel_message(
196 &self,
197 channel_id: ChannelId,
198 user_id: UserId,
199 message_id: MessageId,
200 ) -> Result<()> {
201 self.transaction(|tx| async move {
202 self.observe_channel_message_internal(channel_id, user_id, message_id, &*tx)
203 .await?;
204 Ok(())
205 })
206 .await
207 }
208
209 async fn observe_channel_message_internal(
210 &self,
211 channel_id: ChannelId,
212 user_id: UserId,
213 message_id: MessageId,
214 tx: &DatabaseTransaction,
215 ) -> Result<()> {
216 observed_channel_messages::Entity::insert(observed_channel_messages::ActiveModel {
217 user_id: ActiveValue::Set(user_id),
218 channel_id: ActiveValue::Set(channel_id),
219 channel_message_id: ActiveValue::Set(message_id),
220 })
221 .on_conflict(
222 OnConflict::columns([
223 observed_channel_messages::Column::ChannelId,
224 observed_channel_messages::Column::UserId,
225 ])
226 .update_column(observed_channel_messages::Column::ChannelMessageId)
227 .action_cond_where(observed_channel_messages::Column::ChannelMessageId.lt(message_id))
228 .to_owned(),
229 )
230 // TODO: Try to upgrade SeaORM so we don't have to do this hack around their bug
231 .exec_without_returning(&*tx)
232 .await?;
233 Ok(())
234 }
235
236 pub async fn unseen_channel_messages(
237 &self,
238 user_id: UserId,
239 channel_ids: &[ChannelId],
240 tx: &DatabaseTransaction,
241 ) -> Result<Vec<proto::UnseenChannelMessage>> {
242 let mut observed_messages_by_channel_id = HashMap::default();
243 let mut rows = observed_channel_messages::Entity::find()
244 .filter(observed_channel_messages::Column::UserId.eq(user_id))
245 .filter(observed_channel_messages::Column::ChannelId.is_in(channel_ids.iter().copied()))
246 .stream(&*tx)
247 .await?;
248
249 while let Some(row) = rows.next().await {
250 let row = row?;
251 observed_messages_by_channel_id.insert(row.channel_id, row);
252 }
253 drop(rows);
254 let mut values = String::new();
255 for id in channel_ids {
256 if !values.is_empty() {
257 values.push_str(", ");
258 }
259 write!(&mut values, "({})", id).unwrap();
260 }
261
262 if values.is_empty() {
263 return Ok(Default::default());
264 }
265
266 let sql = format!(
267 r#"
268 SELECT
269 *
270 FROM (
271 SELECT
272 *,
273 row_number() OVER (
274 PARTITION BY channel_id
275 ORDER BY id DESC
276 ) as row_number
277 FROM channel_messages
278 WHERE
279 channel_id in ({values})
280 ) AS messages
281 WHERE
282 row_number = 1
283 "#,
284 );
285
286 let stmt = Statement::from_string(self.pool.get_database_backend(), sql);
287 let last_messages = channel_message::Model::find_by_statement(stmt)
288 .all(&*tx)
289 .await?;
290
291 let mut changes = Vec::new();
292 for last_message in last_messages {
293 if let Some(observed_message) =
294 observed_messages_by_channel_id.get(&last_message.channel_id)
295 {
296 if observed_message.channel_message_id == last_message.id {
297 continue;
298 }
299 }
300 changes.push(proto::UnseenChannelMessage {
301 channel_id: last_message.channel_id.to_proto(),
302 message_id: last_message.id.to_proto(),
303 });
304 }
305
306 Ok(changes)
307 }
308
309 pub async fn remove_channel_message(
310 &self,
311 channel_id: ChannelId,
312 message_id: MessageId,
313 user_id: UserId,
314 ) -> Result<Vec<ConnectionId>> {
315 self.transaction(|tx| async move {
316 let mut rows = channel_chat_participant::Entity::find()
317 .filter(channel_chat_participant::Column::ChannelId.eq(channel_id))
318 .stream(&*tx)
319 .await?;
320
321 let mut is_participant = false;
322 let mut participant_connection_ids = Vec::new();
323 while let Some(row) = rows.next().await {
324 let row = row?;
325 if row.user_id == user_id {
326 is_participant = true;
327 }
328 participant_connection_ids.push(row.connection());
329 }
330 drop(rows);
331
332 if !is_participant {
333 Err(anyhow!("not a chat participant"))?;
334 }
335
336 let result = channel_message::Entity::delete_by_id(message_id)
337 .filter(channel_message::Column::SenderId.eq(user_id))
338 .exec(&*tx)
339 .await?;
340 if result.rows_affected == 0 {
341 Err(anyhow!("no such message"))?;
342 }
343
344 Ok(participant_connection_ids)
345 })
346 .await
347 }
348}