From 6312ff80885c3278863220587104affe8936d05f Mon Sep 17 00:00:00 2001 From: timedout Date: Fri, 26 Jun 2026 00:34:19 +0100 Subject: [PATCH] style: Document & refactor sync service --- src/service/sync/watch.rs | 179 +++++++++++++++++++------------------- 1 file changed, 91 insertions(+), 88 deletions(-) diff --git a/src/service/sync/watch.rs b/src/service/sync/watch.rs index ce8f20f5f..3795b4338 100644 --- a/src/service/sync/watch.rs +++ b/src/service/sync/watch.rs @@ -1,112 +1,115 @@ -use conduwuit::{Result, implement, trace}; +use conduwuit::{Result, trace}; use futures::{FutureExt, StreamExt, pin_mut, stream::FuturesUnordered}; use ruma::{DeviceId, UserId}; -#[implement(super::Service)] -#[tracing::instrument(skip(self), level = "debug")] -pub async fn watch(&self, user_id: &UserId, device_id: &DeviceId) -> Result { - let userid_bytes = user_id.as_bytes().to_vec(); - let mut userid_prefix = userid_bytes.clone(); - userid_prefix.push(0xFF); +impl super::Service { + /// Watches for changes that might wake the sync loop for the given user + + /// device. + #[tracing::instrument(skip(self), level = "debug")] + pub async fn watch(&self, user_id: &UserId, device_id: &DeviceId) -> Result { + let userid_bytes = user_id.as_bytes().to_vec(); + let mut userid_prefix = userid_bytes.clone(); + userid_prefix.push(0xFF); - let mut userdeviceid_prefix = userid_prefix.clone(); - userdeviceid_prefix.extend_from_slice(device_id.as_bytes()); - userdeviceid_prefix.push(0xFF); + let mut userdeviceid_prefix = userid_prefix.clone(); + userdeviceid_prefix.extend_from_slice(device_id.as_bytes()); + userdeviceid_prefix.push(0xFF); - let mut futures = FuturesUnordered::new(); + let mut futures = FuturesUnordered::new(); - // Return when *any* user changed their key - // TODO: only send for user they share a room with - futures.push(self.db.todeviceid_events.watch_prefix(&userdeviceid_prefix)); + // Return when *any* user changed their key + // TODO: only send for user they share a room with + futures.push(self.db.todeviceid_events.watch_prefix(&userdeviceid_prefix)); - futures.push(self.db.userroomid_joined.watch_prefix(&userid_prefix)); - futures.push(self.db.userroomid_invitestate.watch_prefix(&userid_prefix)); - futures.push(self.db.userroomid_leftstate.watch_prefix(&userid_prefix)); - futures.push( - self.db - .userroomid_notificationcount - .watch_prefix(&userid_prefix), - ); - futures.push( - self.db - .userroomid_highlightcount - .watch_prefix(&userid_prefix), - ); + futures.push(self.db.userroomid_joined.watch_prefix(&userid_prefix)); + futures.push(self.db.userroomid_invitestate.watch_prefix(&userid_prefix)); + futures.push(self.db.userroomid_leftstate.watch_prefix(&userid_prefix)); + futures.push( + self.db + .userroomid_notificationcount + .watch_prefix(&userid_prefix), + ); + futures.push( + self.db + .userroomid_highlightcount + .watch_prefix(&userid_prefix), + ); - // Events for rooms we are in - let rooms_joined = self.services.state_cache.rooms_joined(user_id); + // Events for rooms we are in + let rooms_joined = self.services.state_cache.rooms_joined(user_id); - pin_mut!(rooms_joined); - while let Some(room_id) = rooms_joined.next().await { - let Ok(short_roomid) = self.services.short.get_shortroomid(&room_id).await else { - continue; - }; + pin_mut!(rooms_joined); + while let Some(room_id) = rooms_joined.next().await { + let Ok(short_roomid) = self.services.short.get_shortroomid(&room_id).await else { + continue; + }; - let roomid_bytes = room_id.as_bytes().to_vec(); - let mut roomid_prefix = roomid_bytes.clone(); - roomid_prefix.push(0xFF); + let roomid_bytes = room_id.as_bytes().to_vec(); + let mut roomid_prefix = roomid_bytes.clone(); + roomid_prefix.push(0xFF); - // Key changes - futures.push(self.db.keychangeid_userid.watch_prefix(&roomid_prefix)); + // Key changes + futures.push(self.db.keychangeid_userid.watch_prefix(&roomid_prefix)); - // Room account data - let mut roomuser_prefix = roomid_prefix.clone(); - roomuser_prefix.extend_from_slice(&userid_prefix); + // Room account data + let mut roomuser_prefix = roomid_prefix.clone(); + roomuser_prefix.extend_from_slice(&userid_prefix); + + futures.push( + self.db + .roomusertype_roomuserdataid + .watch_prefix(&roomuser_prefix), + ); + + // PDUs + let short_roomid = short_roomid.to_be_bytes().to_vec(); + futures.push(self.db.pduid_pdu.watch_prefix(&short_roomid)); + + // EDUs + let typing_room_id = room_id.clone(); + let typing_wait_for_update = async move { + self.services.typing.wait_for_update(&typing_room_id).await; + }; + + futures.push(typing_wait_for_update.boxed()); + futures.push( + self.db + .readreceiptid_readreceipt + .watch_prefix(&roomid_prefix), + ); + } + + let mut globaluserdata_prefix = vec![0xFF]; + globaluserdata_prefix.extend_from_slice(&userid_prefix); futures.push( self.db .roomusertype_roomuserdataid - .watch_prefix(&roomuser_prefix), + .watch_prefix(&globaluserdata_prefix), ); - // PDUs - let short_roomid = short_roomid.to_be_bytes().to_vec(); - futures.push(self.db.pduid_pdu.watch_prefix(&short_roomid)); + // More key changes (used when user is not joined to any rooms) + futures.push(self.db.keychangeid_userid.watch_prefix(&userid_prefix)); - // EDUs - let typing_room_id = room_id.clone(); - let typing_wait_for_update = async move { - self.services.typing.wait_for_update(&typing_room_id).await; - }; - - futures.push(typing_wait_for_update.boxed()); + // One time keys futures.push( self.db - .readreceiptid_readreceipt - .watch_prefix(&roomid_prefix), + .userid_lastonetimekeyupdate + .watch_prefix(&userid_bytes), ); + + // Server shutdown + futures.push(self.services.server.until_shutdown().boxed()); + + if !self.services.server.running() { + return Ok(()); + } + + // Wait until one of them finds something + trace!(futures = futures.len(), "watch started"); + futures.next().await; + trace!(futures = futures.len(), "watch finished"); + + Ok(()) } - - let mut globaluserdata_prefix = vec![0xFF]; - globaluserdata_prefix.extend_from_slice(&userid_prefix); - - futures.push( - self.db - .roomusertype_roomuserdataid - .watch_prefix(&globaluserdata_prefix), - ); - - // More key changes (used when user is not joined to any rooms) - futures.push(self.db.keychangeid_userid.watch_prefix(&userid_prefix)); - - // One time keys - futures.push( - self.db - .userid_lastonetimekeyupdate - .watch_prefix(&userid_bytes), - ); - - // Server shutdown - futures.push(self.services.server.until_shutdown().boxed()); - - if !self.services.server.running() { - return Ok(()); - } - - // Wait until one of them finds something - trace!(futures = futures.len(), "watch started"); - futures.next().await; - trace!(futures = futures.len(), "watch finished"); - - Ok(()) }