From d2817679e59494a7944e1141e65dc236bcf7b061 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Timo=20K=C3=B6sters?= Date: Fri, 22 Mar 2024 08:22:15 +0100 Subject: [PATCH] refactor: remove previous typing implementation and add sync wakeup for new one --- src/api/client_server/sync.rs | 9 +- src/api/client_server/typing.rs | 18 ++- src/api/server_server.rs | 18 ++- src/database/key_value/globals.rs | 4 +- src/database/key_value/rooms/edus/mod.rs | 1 - src/database/key_value/rooms/edus/typing.rs | 127 -------------------- src/database/mod.rs | 4 - src/service/mod.rs | 8 +- src/service/rooms/edus/mod.rs | 2 +- src/service/rooms/edus/typing/data.rs | 21 ---- src/service/rooms/edus/typing/mod.rs | 74 +++++++++--- 11 files changed, 101 insertions(+), 185 deletions(-) delete mode 100644 src/database/key_value/rooms/edus/typing.rs delete mode 100644 src/service/rooms/edus/typing/data.rs diff --git a/src/api/client_server/sync.rs b/src/api/client_server/sync.rs index ba2ceefa..da603dcf 100644 --- a/src/api/client_server/sync.rs +++ b/src/api/client_server/sync.rs @@ -1055,7 +1055,14 @@ async fn load_joined_room( .map(|(_, _, v)| v) .collect(); - if services().rooms.edus.typing.last_typing_update(room_id).await? > since { + if services() + .rooms + .edus + .typing + .last_typing_update(room_id) + .await? + > since + { edus.push( serde_json::from_str( &serde_json::to_string(&services().rooms.edus.typing.typings_all(room_id).await?) diff --git a/src/api/client_server/typing.rs b/src/api/client_server/typing.rs index eff84051..e9e93708 100644 --- a/src/api/client_server/typing.rs +++ b/src/api/client_server/typing.rs @@ -23,17 +23,23 @@ pub async fn create_typing_event_route( } if let Typing::Yes(duration) = body.state { - services().rooms.edus.typing.typing_add( - sender_user, - &body.room_id, - duration.as_millis() as u64 + utils::millis_since_unix_epoch(), - ).await?; + services() + .rooms + .edus + .typing + .typing_add( + sender_user, + &body.room_id, + duration.as_millis() as u64 + utils::millis_since_unix_epoch(), + ) + .await?; } else { services() .rooms .edus .typing - .typing_remove(sender_user, &body.room_id).await?; + .typing_remove(sender_user, &body.room_id) + .await?; } Ok(create_typing_event::v3::Response {}) diff --git a/src/api/server_server.rs b/src/api/server_server.rs index 59724b03..0fdf22fc 100644 --- a/src/api/server_server.rs +++ b/src/api/server_server.rs @@ -826,17 +826,23 @@ pub async fn send_transaction_message_route( .is_joined(&typing.user_id, &typing.room_id)? { if typing.typing { - services().rooms.edus.typing.typing_add( - &typing.user_id, - &typing.room_id, - 3000 + utils::millis_since_unix_epoch(), - ).await?; + services() + .rooms + .edus + .typing + .typing_add( + &typing.user_id, + &typing.room_id, + 3000 + utils::millis_since_unix_epoch(), + ) + .await?; } else { services() .rooms .edus .typing - .typing_remove(&typing.user_id, &typing.room_id).await?; + .typing_remove(&typing.user_id, &typing.room_id) + .await?; } } } diff --git a/src/database/key_value/globals.rs b/src/database/key_value/globals.rs index 7a70af18..2851ce53 100644 --- a/src/database/key_value/globals.rs +++ b/src/database/key_value/globals.rs @@ -94,7 +94,9 @@ impl service::globals::Data for KeyValueDatabase { futures.push(self.pduid_pdu.watch_prefix(&short_roomid)); // EDUs - futures.push(self.roomid_lasttypingupdate.watch_prefix(&roomid_bytes)); + futures.push(Box::into_pin(Box::new(async move { + let _result = services().rooms.edus.typing.wait_for_update(&room_id).await; + }))); futures.push(self.readreceiptid_readreceipt.watch_prefix(&roomid_prefix)); diff --git a/src/database/key_value/rooms/edus/mod.rs b/src/database/key_value/rooms/edus/mod.rs index 6c652918..7abf946f 100644 --- a/src/database/key_value/rooms/edus/mod.rs +++ b/src/database/key_value/rooms/edus/mod.rs @@ -1,6 +1,5 @@ mod presence; mod read_receipt; -mod typing; use crate::{database::KeyValueDatabase, service}; diff --git a/src/database/key_value/rooms/edus/typing.rs b/src/database/key_value/rooms/edus/typing.rs deleted file mode 100644 index 5709192c..00000000 --- a/src/database/key_value/rooms/edus/typing.rs +++ /dev/null @@ -1,127 +0,0 @@ -use std::{collections::HashSet, mem}; - -use ruma::{OwnedUserId, RoomId, UserId}; - -use crate::{database::KeyValueDatabase, service, services, utils, Error, Result}; - -impl service::rooms::edus::typing::Data for KeyValueDatabase { - fn typing_add(&self, user_id: &UserId, room_id: &RoomId, timeout: u64) -> Result<()> { - let mut prefix = room_id.as_bytes().to_vec(); - prefix.push(0xff); - - let count = services().globals.next_count()?.to_be_bytes(); - - let mut room_typing_id = prefix; - room_typing_id.extend_from_slice(&timeout.to_be_bytes()); - room_typing_id.push(0xff); - room_typing_id.extend_from_slice(&count); - - self.typingid_userid - .insert(&room_typing_id, user_id.as_bytes())?; - - self.roomid_lasttypingupdate - .insert(room_id.as_bytes(), &count)?; - - Ok(()) - } - - fn typing_remove(&self, user_id: &UserId, room_id: &RoomId) -> Result<()> { - let mut prefix = room_id.as_bytes().to_vec(); - prefix.push(0xff); - - let user_id = user_id.to_string(); - - let mut found_outdated = false; - - // Maybe there are multiple ones from calling roomtyping_add multiple times - for outdated_edu in self - .typingid_userid - .scan_prefix(prefix) - .filter(|(_, v)| &**v == user_id.as_bytes()) - { - self.typingid_userid.remove(&outdated_edu.0)?; - found_outdated = true; - } - - if found_outdated { - self.roomid_lasttypingupdate.insert( - room_id.as_bytes(), - &services().globals.next_count()?.to_be_bytes(), - )?; - } - - Ok(()) - } - - fn typings_maintain(&self, room_id: &RoomId) -> Result<()> { - let mut prefix = room_id.as_bytes().to_vec(); - prefix.push(0xff); - - let current_timestamp = utils::millis_since_unix_epoch(); - - let mut found_outdated = false; - - // Find all outdated edus before inserting a new one - for outdated_edu in self - .typingid_userid - .scan_prefix(prefix) - .map(|(key, _)| { - Ok::<_, Error>(( - key.clone(), - utils::u64_from_bytes( - &key.splitn(2, |&b| b == 0xff).nth(1).ok_or_else(|| { - Error::bad_database("RoomTyping has invalid timestamp or delimiters.") - })?[0..mem::size_of::()], - ) - .map_err(|_| Error::bad_database("RoomTyping has invalid timestamp bytes."))?, - )) - }) - .filter_map(|r| r.ok()) - .take_while(|&(_, timestamp)| timestamp < current_timestamp) - { - // This is an outdated edu (time > timestamp) - self.typingid_userid.remove(&outdated_edu.0)?; - found_outdated = true; - } - - if found_outdated { - self.roomid_lasttypingupdate.insert( - room_id.as_bytes(), - &services().globals.next_count()?.to_be_bytes(), - )?; - } - - Ok(()) - } - - fn last_typing_update(&self, room_id: &RoomId) -> Result { - Ok(self - .roomid_lasttypingupdate - .get(room_id.as_bytes())? - .map(|bytes| { - utils::u64_from_bytes(&bytes).map_err(|_| { - Error::bad_database("Count in roomid_lastroomactiveupdate is invalid.") - }) - }) - .transpose()? - .unwrap_or(0)) - } - - fn typings_all(&self, room_id: &RoomId) -> Result> { - let mut prefix = room_id.as_bytes().to_vec(); - prefix.push(0xff); - - let mut user_ids = HashSet::new(); - - for (_, user_id) in self.typingid_userid.scan_prefix(prefix) { - let user_id = UserId::parse(utils::string_from_bytes(&user_id).map_err(|_| { - Error::bad_database("User ID in typingid_userid is invalid unicode.") - })?) - .map_err(|_| Error::bad_database("User ID in typingid_userid is invalid."))?; - - user_ids.insert(user_id); - } - - Ok(user_ids) - } -} diff --git a/src/database/mod.rs b/src/database/mod.rs index 425ef4e9..0960dc96 100644 --- a/src/database/mod.rs +++ b/src/database/mod.rs @@ -70,8 +70,6 @@ pub struct KeyValueDatabase { pub(super) readreceiptid_readreceipt: Arc, // ReadReceiptId = RoomId + Count + UserId pub(super) roomuserid_privateread: Arc, // RoomUserId = Room + User, PrivateRead = Count pub(super) roomuserid_lastprivatereadupdate: Arc, // LastPrivateReadUpdate = Count - pub(super) typingid_userid: Arc, // TypingId = RoomId + TimeoutTime + Count - pub(super) roomid_lasttypingupdate: Arc, // LastRoomTypingUpdate = Count pub(super) presenceid_presence: Arc, // PresenceId = RoomId + Count + UserId pub(super) userid_lastpresenceupdate: Arc, // LastPresenceUpdate = Count @@ -301,8 +299,6 @@ impl KeyValueDatabase { roomuserid_privateread: builder.open_tree("roomuserid_privateread")?, // "Private" read receipt roomuserid_lastprivatereadupdate: builder .open_tree("roomuserid_lastprivatereadupdate")?, - typingid_userid: builder.open_tree("typingid_userid")?, - roomid_lasttypingupdate: builder.open_tree("roomid_lasttypingupdate")?, presenceid_presence: builder.open_tree("presenceid_presence")?, userid_lastpresenceupdate: builder.open_tree("userid_lastpresenceupdate")?, pduid_pdu: builder.open_tree("pduid_pdu")?, diff --git a/src/service/mod.rs b/src/service/mod.rs index 8c97cdd0..ba1ae1c8 100644 --- a/src/service/mod.rs +++ b/src/service/mod.rs @@ -4,7 +4,7 @@ use std::{ }; use lru_cache::LruCache; -use tokio::sync::Mutex; +use tokio::sync::{broadcast, Mutex}; use crate::{Config, Result}; use tokio::sync::RwLock; @@ -66,7 +66,11 @@ impl Services { edus: rooms::edus::Service { presence: rooms::edus::presence::Service { db }, read_receipt: rooms::edus::read_receipt::Service { db }, - typing: rooms::edus::typing::Service { db, typing: RwLock::new(BTreeMap::new()), last_typing_update: RwLock::new(BTreeMap::new()) }, + typing: rooms::edus::typing::Service { + typing: RwLock::new(BTreeMap::new()), + last_typing_update: RwLock::new(BTreeMap::new()), + typing_update_sender: broadcast::channel(100).0, + }, }, event_handler: rooms::event_handler::Service, lazy_loading: rooms::lazy_loading::Service { diff --git a/src/service/rooms/edus/mod.rs b/src/service/rooms/edus/mod.rs index cf7a3591..a6bc3d5b 100644 --- a/src/service/rooms/edus/mod.rs +++ b/src/service/rooms/edus/mod.rs @@ -2,7 +2,7 @@ pub mod presence; pub mod read_receipt; pub mod typing; -pub trait Data: presence::Data + read_receipt::Data + typing::Data + 'static {} +pub trait Data: presence::Data + read_receipt::Data + 'static {} pub struct Service { pub presence: presence::Service, diff --git a/src/service/rooms/edus/typing/data.rs b/src/service/rooms/edus/typing/data.rs deleted file mode 100644 index 3b1eecfb..00000000 --- a/src/service/rooms/edus/typing/data.rs +++ /dev/null @@ -1,21 +0,0 @@ -use crate::Result; -use ruma::{OwnedUserId, RoomId, UserId}; -use std::collections::HashSet; - -pub trait Data: Send + Sync { - /// Sets a user as typing until the timeout timestamp is reached or roomtyping_remove is - /// called. - fn typing_add(&self, user_id: &UserId, room_id: &RoomId, timeout: u64) -> Result<()>; - - /// Removes a user from typing before the timeout is reached. - fn typing_remove(&self, user_id: &UserId, room_id: &RoomId) -> Result<()>; - - /// Makes sure that typing events with old timestamps get removed. - fn typings_maintain(&self, room_id: &RoomId) -> Result<()>; - - /// Returns the count of the last typing update in this room. - fn last_typing_update(&self, room_id: &RoomId) -> Result; - - /// Returns all user ids currently typing. - fn typings_all(&self, room_id: &RoomId) -> Result>; -} diff --git a/src/service/rooms/edus/typing/mod.rs b/src/service/rooms/edus/typing/mod.rs index f3436741..7546aa84 100644 --- a/src/service/rooms/edus/typing/mod.rs +++ b/src/service/rooms/edus/typing/mod.rs @@ -1,31 +1,57 @@ -mod data; - -pub use data::Data; -use ruma::{events::SyncEphemeralRoomEvent, RoomId, UserId, OwnedRoomId, OwnedUserId}; -use tokio::sync::RwLock; +use ruma::{events::SyncEphemeralRoomEvent, OwnedRoomId, OwnedUserId, RoomId, UserId}; use std::collections::BTreeMap; +use tokio::sync::{broadcast, RwLock}; -use crate::{utils, services, Result}; +use crate::{services, utils, Result}; pub struct Service { - pub db: &'static dyn Data, pub typing: RwLock>>, // u64 is unix timestamp of timeout pub last_typing_update: RwLock>, // timestamp of the last change to typing users + pub typing_update_sender: broadcast::Sender, } impl Service { /// Sets a user as typing until the timeout timestamp is reached or roomtyping_remove is /// called. pub async fn typing_add(&self, user_id: &UserId, room_id: &RoomId, timeout: u64) -> Result<()> { - self.typing.write().await.entry(room_id.to_owned()).or_default().insert(user_id.to_owned(), timeout); - self.last_typing_update.write().await.insert(room_id.to_owned(), services().globals.next_count()?); + self.typing + .write() + .await + .entry(room_id.to_owned()) + .or_default() + .insert(user_id.to_owned(), timeout); + self.last_typing_update + .write() + .await + .insert(room_id.to_owned(), services().globals.next_count()?); + let _ = self.typing_update_sender.send(room_id.to_owned()); Ok(()) } /// Removes a user from typing before the timeout is reached. pub async fn typing_remove(&self, user_id: &UserId, room_id: &RoomId) -> Result<()> { - self.typing.write().await.entry(room_id.to_owned()).or_default().remove(user_id); - self.last_typing_update.write().await.insert(room_id.to_owned(), services().globals.next_count()?); + self.typing + .write() + .await + .entry(room_id.to_owned()) + .or_default() + .remove(user_id); + self.last_typing_update + .write() + .await + .insert(room_id.to_owned(), services().globals.next_count()?); + let _ = self.typing_update_sender.send(room_id.to_owned()); + Ok(()) + } + + pub async fn wait_for_update(&self, room_id: &RoomId) -> Result<()> { + let mut receiver = self.typing_update_sender.subscribe(); + while let Ok(next) = receiver.recv().await { + if next == room_id { + break; + } + } + Ok(()) } @@ -35,7 +61,9 @@ impl Service { let mut removable = Vec::new(); { let typing = self.typing.read().await; - let Some(room) = typing.get(room_id) else { return Ok(()); }; + let Some(room) = typing.get(room_id) else { + return Ok(()); + }; for (user, timeout) in room { if *timeout < current_timestamp { removable.push(user.clone()); @@ -49,7 +77,11 @@ impl Service { for user in removable { room.remove(&user); } - self.last_typing_update.write().await.insert(room_id.to_owned(), services().globals.next_count()?); + self.last_typing_update + .write() + .await + .insert(room_id.to_owned(), services().globals.next_count()?); + let _ = self.typing_update_sender.send(room_id.to_owned()); } Ok(()) } @@ -57,7 +89,13 @@ impl Service { /// Returns the count of the last typing update in this room. pub async fn last_typing_update(&self, room_id: &RoomId) -> Result { self.typings_maintain(room_id).await?; - Ok(self.last_typing_update.read().await.get(room_id).copied().unwrap_or(0)) + Ok(self + .last_typing_update + .read() + .await + .get(room_id) + .copied() + .unwrap_or(0)) } /// Returns a new typing EDU. @@ -67,7 +105,13 @@ impl Service { ) -> Result> { Ok(SyncEphemeralRoomEvent { content: ruma::events::typing::TypingEventContent { - user_ids: self.typing.read().await.get(room_id).map(|m| m.keys().cloned().collect()).unwrap_or_default(), + user_ids: self + .typing + .read() + .await + .get(room_id) + .map(|m| m.keys().cloned().collect()) + .unwrap_or_default(), }, }) }