Remove channel chat participant when connection is lost

crates/collab/src/db/queries/buffers.rs

@@ -249,6 +249,29 @@ impl Database {
         .await
     }
 
+    pub async fn channel_buffer_connection_lost(
+        &self,
+        connection: ConnectionId,
+        tx: &DatabaseTransaction,
+    ) -> Result<()> {
+        channel_buffer_collaborator::Entity::update_many()
+            .filter(
+                Condition::all()
+                    .add(channel_buffer_collaborator::Column::ConnectionId.eq(connection.id as i32))
+                    .add(
+                        channel_buffer_collaborator::Column::ConnectionServerId
+                            .eq(connection.owner_id as i32),
+                    ),
+            )
+            .set(channel_buffer_collaborator::ActiveModel {
+                connection_lost: ActiveValue::set(true),
+                ..Default::default()
+            })
+            .exec(&*tx)
+            .await?;
+        Ok(())
+    }
+
     pub async fn leave_channel_buffers(
         &self,
         connection: ConnectionId,

crates/collab/src/db/queries/messages.rs

@@ -25,6 +25,25 @@ impl Database {
         .await
     }
 
+    pub async fn channel_chat_connection_lost(
+        &self,
+        connection_id: ConnectionId,
+        tx: &DatabaseTransaction,
+    ) -> Result<()> {
+        channel_chat_participant::Entity::delete_many()
+            .filter(
+                Condition::all()
+                    .add(
+                        channel_chat_participant::Column::ConnectionServerId
+                            .eq(connection_id.owner_id),
+                    )
+                    .add(channel_chat_participant::Column::ConnectionId.eq(connection_id.id)),
+            )
+            .exec(tx)
+            .await?;
+        Ok(())
+    }
+
     pub async fn leave_channel_chat(
         &self,
         channel_id: ChannelId,

crates/collab/src/db/queries/rooms.rs

@@ -890,54 +890,43 @@ impl Database {
 
     pub async fn connection_lost(&self, connection: ConnectionId) -> Result<()> {
         self.transaction(|tx| async move {
-            let participant = room_participant::Entity::find()
+            self.room_connection_lost(connection, &*tx).await?;
-                .filter(
+            self.channel_buffer_connection_lost(connection, &*tx)
-                    Condition::all()
-                        .add(
-                            room_participant::Column::AnsweringConnectionId
-                                .eq(connection.id as i32),
-                        )
-                        .add(
-                            room_participant::Column::AnsweringConnectionServerId
-                                .eq(connection.owner_id as i32),
-                        ),
-                )
-                .one(&*tx)
                 .await?;
-
+            self.channel_chat_connection_lost(connection, &*tx).await?;
-            if let Some(participant) = participant {
-                room_participant::Entity::update(room_participant::ActiveModel {
-                    answering_connection_lost: ActiveValue::set(true),
-                    ..participant.into_active_model()
-                })
-                .exec(&*tx)
-                .await?;
-            }
-
-            channel_buffer_collaborator::Entity::update_many()
-                .filter(
-                    Condition::all()
-                        .add(
-                            channel_buffer_collaborator::Column::ConnectionId
-                                .eq(connection.id as i32),
-                        )
-                        .add(
-                            channel_buffer_collaborator::Column::ConnectionServerId
-                                .eq(connection.owner_id as i32),
-                        ),
-                )
-                .set(channel_buffer_collaborator::ActiveModel {
-                    connection_lost: ActiveValue::set(true),
-                    ..Default::default()
-                })
-                .exec(&*tx)
-                .await?;
-
             Ok(())
         })
         .await
     }
 
+    pub async fn room_connection_lost(
+        &self,
+        connection: ConnectionId,
+        tx: &DatabaseTransaction,
+    ) -> Result<()> {
+        let participant = room_participant::Entity::find()
+            .filter(
+                Condition::all()
+                    .add(room_participant::Column::AnsweringConnectionId.eq(connection.id as i32))
+                    .add(
+                        room_participant::Column::AnsweringConnectionServerId
+                            .eq(connection.owner_id as i32),
+                    ),
+            )
+            .one(&*tx)
+            .await?;
+
+        if let Some(participant) = participant {
+            room_participant::Entity::update(room_participant::ActiveModel {
+                answering_connection_lost: ActiveValue::set(true),
+                ..participant.into_active_model()
+            })
+            .exec(&*tx)
+            .await?;
+        }
+        Ok(())
+    }
+
     fn build_incoming_call(
         room: &proto::Room,
         called_user_id: UserId,

crates/collab/src/rpc.rs

@@ -904,9 +904,8 @@ async fn connection_lost(
                     room_updated(&room, &session.peer);
                 }
             }
+
             update_user_contacts(session.user_id, &session).await?;
-
-
         }
         _ = teardown.changed().fuse() => {}
     }