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_asc(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 Ok(messages)
114 })
115 .await
116 }
117
118 pub async fn create_channel_message(
119 &self,
120 channel_id: ChannelId,
121 user_id: UserId,
122 body: &str,
123 timestamp: OffsetDateTime,
124 nonce: u128,
125 ) -> Result<(MessageId, Vec<ConnectionId>, Vec<UserId>)> {
126 self.transaction(|tx| async move {
127 let mut rows = channel_chat_participant::Entity::find()
128 .filter(channel_chat_participant::Column::ChannelId.eq(channel_id))
129 .stream(&*tx)
130 .await?;
131
132 let mut is_participant = false;
133 let mut participant_connection_ids = Vec::new();
134 let mut participant_user_ids = Vec::new();
135 while let Some(row) = rows.next().await {
136 let row = row?;
137 if row.user_id == user_id {
138 is_participant = true;
139 }
140 participant_user_ids.push(row.user_id);
141 participant_connection_ids.push(row.connection());
142 }
143 drop(rows);
144
145 if !is_participant {
146 Err(anyhow!("not a chat participant"))?;
147 }
148
149 let timestamp = timestamp.to_offset(time::UtcOffset::UTC);
150 let timestamp = time::PrimitiveDateTime::new(timestamp.date(), timestamp.time());
151
152 let message = channel_message::Entity::insert(channel_message::ActiveModel {
153 channel_id: ActiveValue::Set(channel_id),
154 sender_id: ActiveValue::Set(user_id),
155 body: ActiveValue::Set(body.to_string()),
156 sent_at: ActiveValue::Set(timestamp),
157 nonce: ActiveValue::Set(Uuid::from_u128(nonce)),
158 id: ActiveValue::NotSet,
159 })
160 .on_conflict(
161 OnConflict::column(channel_message::Column::Nonce)
162 .update_column(channel_message::Column::Nonce)
163 .to_owned(),
164 )
165 .exec(&*tx)
166 .await?;
167
168 #[derive(Debug, Clone, Copy, EnumIter, DeriveColumn)]
169 enum QueryConnectionId {
170 ConnectionId,
171 }
172
173 // Observe this message for the sender
174 self.observe_channel_message_internal(
175 channel_id,
176 user_id,
177 message.last_insert_id,
178 &*tx,
179 )
180 .await?;
181
182 let mut channel_members = self.get_channel_members_internal(channel_id, &*tx).await?;
183 channel_members.retain(|member| !participant_user_ids.contains(member));
184
185 Ok((
186 message.last_insert_id,
187 participant_connection_ids,
188 channel_members,
189 ))
190 })
191 .await
192 }
193
194 pub async fn observe_channel_message(
195 &self,
196 channel_id: ChannelId,
197 user_id: UserId,
198 message_id: MessageId,
199 ) -> Result<()> {
200 self.transaction(|tx| async move {
201 self.observe_channel_message_internal(channel_id, user_id, message_id, &*tx)
202 .await?;
203 Ok(())
204 })
205 .await
206 }
207
208 async fn observe_channel_message_internal(
209 &self,
210 channel_id: ChannelId,
211 user_id: UserId,
212 message_id: MessageId,
213 tx: &DatabaseTransaction,
214 ) -> Result<()> {
215 observed_channel_messages::Entity::insert(observed_channel_messages::ActiveModel {
216 user_id: ActiveValue::Set(user_id),
217 channel_id: ActiveValue::Set(channel_id),
218 channel_message_id: ActiveValue::Set(message_id),
219 })
220 .on_conflict(
221 OnConflict::columns([
222 observed_channel_messages::Column::ChannelId,
223 observed_channel_messages::Column::UserId,
224 ])
225 .update_column(observed_channel_messages::Column::ChannelMessageId)
226 .action_cond_where(observed_channel_messages::Column::ChannelMessageId.lt(message_id))
227 .to_owned(),
228 )
229 // TODO: Try to upgrade SeaORM so we don't have to do this hack around their bug
230 .exec_without_returning(&*tx)
231 .await?;
232 Ok(())
233 }
234
235 pub async fn unseen_channel_messages(
236 &self,
237 user_id: UserId,
238 channel_ids: &[ChannelId],
239 tx: &DatabaseTransaction,
240 ) -> Result<Vec<proto::UnseenChannelMessage>> {
241 let mut observed_messages_by_channel_id = HashMap::default();
242 let mut rows = observed_channel_messages::Entity::find()
243 .filter(observed_channel_messages::Column::UserId.eq(user_id))
244 .filter(observed_channel_messages::Column::ChannelId.is_in(channel_ids.iter().copied()))
245 .stream(&*tx)
246 .await?;
247
248 while let Some(row) = rows.next().await {
249 let row = row?;
250 observed_messages_by_channel_id.insert(row.channel_id, row);
251 }
252 drop(rows);
253 let mut values = String::new();
254 for id in channel_ids {
255 if !values.is_empty() {
256 values.push_str(", ");
257 }
258 write!(&mut values, "({})", id).unwrap();
259 }
260
261 if values.is_empty() {
262 return Ok(Default::default());
263 }
264
265 let sql = format!(
266 r#"
267 SELECT
268 *
269 FROM (
270 SELECT
271 *,
272 row_number() OVER (
273 PARTITION BY channel_id
274 ORDER BY id DESC
275 ) as row_number
276 FROM channel_messages
277 WHERE
278 channel_id in ({values})
279 ) AS messages
280 WHERE
281 row_number = 1
282 "#,
283 );
284
285 let stmt = Statement::from_string(self.pool.get_database_backend(), sql);
286 let last_messages = channel_message::Model::find_by_statement(stmt)
287 .all(&*tx)
288 .await?;
289
290 let mut changes = Vec::new();
291 for last_message in last_messages {
292 if let Some(observed_message) =
293 observed_messages_by_channel_id.get(&last_message.channel_id)
294 {
295 if observed_message.channel_message_id == last_message.id {
296 continue;
297 }
298 }
299 changes.push(proto::UnseenChannelMessage {
300 channel_id: last_message.channel_id.to_proto(),
301 message_id: last_message.id.to_proto(),
302 });
303 }
304
305 Ok(changes)
306 }
307
308 pub async fn remove_channel_message(
309 &self,
310 channel_id: ChannelId,
311 message_id: MessageId,
312 user_id: UserId,
313 ) -> Result<Vec<ConnectionId>> {
314 self.transaction(|tx| async move {
315 let mut rows = channel_chat_participant::Entity::find()
316 .filter(channel_chat_participant::Column::ChannelId.eq(channel_id))
317 .stream(&*tx)
318 .await?;
319
320 let mut is_participant = false;
321 let mut participant_connection_ids = Vec::new();
322 while let Some(row) = rows.next().await {
323 let row = row?;
324 if row.user_id == user_id {
325 is_participant = true;
326 }
327 participant_connection_ids.push(row.connection());
328 }
329 drop(rows);
330
331 if !is_participant {
332 Err(anyhow!("not a chat participant"))?;
333 }
334
335 let result = channel_message::Entity::delete_by_id(message_id)
336 .filter(channel_message::Column::SenderId.eq(user_id))
337 .exec(&*tx)
338 .await?;
339 if result.rows_affected == 0 {
340 Err(anyhow!("no such message"))?;
341 }
342
343 Ok(participant_connection_ids)
344 })
345 .await
346 }
347}