Use ON CONFLICT and transactions where appropriate.

src/db/sqlite/rooms.rs

@@ -51,16 +51,13 @@ impl Rooms for Database {
     }
 
     async fn insert_room_info(&self, info: &RoomInfo) -> Result<(), DataError> {
-        //Clear out old info first, because we want this to be an "upsert."
+        sqlx::query(
-        sqlx::query("DELETE FROM room_info where room_id = ?")
+            r#"INSERT INTO room_info (room_id, room_name) VALUES (?, ?)
-            .bind(&info.room_id)
+               ON CONFLICT(room_id) DO UPDATE SET room_name = ?"#,
-            .execute(&self.conn)
+        )
-            .await
-            .ok();
-
-        sqlx::query(r#"INSERT INTO room_info (room_id, room_name) VALUES (?, ?)"#)
         .bind(&info.room_id)
         .bind(&info.room_name)
+        .bind(&info.room_name)
         .execute(&self.conn)
         .await?;
 
@@ -107,12 +104,10 @@ impl Rooms for Database {
     }
 
     async fn add_user_to_room(&self, username: &str, room_id: &str) -> Result<(), DataError> {
-        // This is here because it is possible to process a bunch of
+        sqlx::query(
-        // user join/leave events at once, and we don't want to cause
+            "INSERT INTO room_users (room_id, username) VALUES (?, ?)
-        // constraint violation errors.
+             ON CONFLICT DO NOTHING",
-        self.remove_user_from_room(username, room_id).await.ok();
+        )
-
-        sqlx::query("INSERT INTO room_users (room_id, username) VALUES (?, ?)")
         .bind(room_id)
         .bind(username)
         .execute(&self.conn)
@@ -134,16 +129,19 @@ impl Rooms for Database {
     async fn clear_info(&self, room_id: &str) -> Result<(), DataError> {
         // We do not clear event history here, because if we rejoin a
         // room, we would re-process events we've already seen.
+        let mut tx = self.conn.begin().await?;
+
         sqlx::query("DELETE FROM room_info where room_id = ?")
             .bind(room_id)
-            .execute(&self.conn)
+            .execute(&mut tx)
             .await?;
 
         sqlx::query("DELETE FROM room_users where room_id = ?")
             .bind(room_id)
-            .execute(&self.conn)
+            .execute(&mut tx)
             .await?;
 
+        tx.commit().await?;
         Ok(())
     }
 }