magmaus3/continuwuity
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Add initial MSC4186 (Simplified Sliding Sync) implementation

Browse source 
Signed-off-by: morguldir <morguldir@protonmail.com>
This commit is contained in:
morguldir 2025-01-03 08:32:54 +01:00 • committed by strawberry
parent 099c9fb22f
commit 721659f22a
7 changed files with 1197 additions and 92 deletions
Download patch file Download diff file Expand all files Collapse all files

273
src/service/sync/mod.rs
View file

@ -11,8 +11,9 @@ use ruma::{
api::client::sync::sync_events::{
self,
v4::{ExtensionsConfig, SyncRequestList},
v5,
},
OwnedDeviceId, OwnedRoomId, OwnedUserId,
DeviceId, OwnedDeviceId, OwnedRoomId, OwnedUserId, UserId,
};
use crate::{rooms, Dep};
@ -20,7 +21,8 @@ use crate::{rooms, Dep};
pub struct Service {
db: Data,
services: Services,
connections: DbConnections,
connections: DbConnections<DbConnectionsKey, DbConnectionsVal>,
snake_connections: DbConnections<SnakeConnectionsKey, SnakeConnectionsVal>,
}
pub struct Data {
@ -52,9 +54,19 @@ struct SlidingSyncCache {
extensions: ExtensionsConfig,
}
type DbConnections = Mutex<BTreeMap<DbConnectionsKey, DbConnectionsVal>>;
#[derive(Default)]
struct SnakeSyncCache {
lists: BTreeMap<String, v5::request::List>,
subscriptions: BTreeMap<OwnedRoomId, v5::request::RoomSubscription>,
known_rooms: BTreeMap<String, BTreeMap<OwnedRoomId, u64>>,
extensions: v5::request::Extensions,
}
type DbConnections<K, V> = Mutex<BTreeMap<K, V>>;
type DbConnectionsKey = (OwnedUserId, OwnedDeviceId, String);
type DbConnectionsVal = Arc<Mutex<SlidingSyncCache>>;
type SnakeConnectionsKey = (OwnedUserId, OwnedDeviceId, Option<String>);
type SnakeConnectionsVal = Arc<Mutex<SnakeSyncCache>>;
impl crate::Service for Service {
fn build(args: crate::Args<'_>) -> Result<Arc<Self>> {
@ -79,12 +91,15 @@ impl crate::Service for Service {
typing: args.depend::<rooms::typing::Service>("rooms::typing"),
},
connections: StdMutex::new(BTreeMap::new()),
snake_connections: StdMutex::new(BTreeMap::new()),
}))
}
fn name(&self) -> &str { crate::service::make_name(std::module_path!()) }
}
/// load params from cache if body doesn't contain it, as long as it's allowed
// in some cases we may need to allow an empty list as an actual value
fn list_or_sticky<T: Clone>(target: &mut Vec<T>, cached: &Vec<T>) {
if target.is_empty() {
target.clone_from(cached);
@ -97,6 +112,30 @@ fn some_or_sticky<T>(target: &mut Option<T>, cached: Option<T>) {
}
impl Service {
pub fn snake_connection_cached(
&self,
user_id: OwnedUserId,
device_id: OwnedDeviceId,
conn_id: Option<String>,
) -> bool {
self.snake_connections
.lock()
.unwrap()
.contains_key(&(user_id, device_id, conn_id))
}
pub fn forget_snake_sync_connection(
&self,
user_id: OwnedUserId,
device_id: OwnedDeviceId,
conn_id: Option<String>,
) {
self.snake_connections
.lock()
.expect("locked")
.remove(&(user_id, device_id, conn_id));
}
pub fn remembered(
&self,
user_id: OwnedUserId,
@ -121,6 +160,112 @@ impl Service {
.remove(&(user_id, device_id, conn_id));
}
pub fn update_snake_sync_request_with_cache(
&self,
user_id: OwnedUserId,
device_id: OwnedDeviceId,
request: &mut v5::Request,
) -> BTreeMap<String, BTreeMap<OwnedRoomId, u64>> {
let conn_id = request.conn_id.clone();
let mut cache = self.snake_connections.lock().expect("locked");
let cached = Arc::clone(
cache
.entry((user_id, device_id, conn_id))
.or_insert_with(|| Arc::new(Mutex::new(SnakeSyncCache::default()))),
);
let cached = &mut cached.lock().expect("locked");
drop(cache);
//v5::Request::try_from_http_request(req, path_args);
for (list_id, list) in &mut request.lists {
if let Some(cached_list) = cached.lists.get(list_id) {
list_or_sticky(
&mut list.room_details.required_state,
&cached_list.room_details.required_state,
);
some_or_sticky(&mut list.include_heroes, cached_list.include_heroes);
match (&mut list.filters, cached_list.filters.clone()) {
| (Some(filters), Some(cached_filters)) => {
some_or_sticky(&mut filters.is_invite, cached_filters.is_invite);
// TODO (morguldir): Find out how a client can unset this, probably need
// to change into an option inside ruma
list_or_sticky(
&mut filters.not_room_types,
&cached_filters.not_room_types,
);
},
| (_, Some(cached_filters)) => list.filters = Some(cached_filters),
| (Some(list_filters), _) => list.filters = Some(list_filters.clone()),
| (..) => {},
}
}
cached.lists.insert(list_id.clone(), list.clone());
}
cached
.subscriptions
.extend(request.room_subscriptions.clone());
request
.room_subscriptions
.extend(cached.subscriptions.clone());
request.extensions.e2ee.enabled = request
.extensions
.e2ee
.enabled
.or(cached.extensions.e2ee.enabled);
request.extensions.to_device.enabled = request
.extensions
.to_device
.enabled
.or(cached.extensions.to_device.enabled);
request.extensions.account_data.enabled = request
.extensions
.account_data
.enabled
.or(cached.extensions.account_data.enabled);
request.extensions.account_data.lists = request
.extensions
.account_data
.lists
.clone()
.or_else(|| cached.extensions.account_data.lists.clone());
request.extensions.account_data.rooms = request
.extensions
.account_data
.rooms
.clone()
.or_else(|| cached.extensions.account_data.rooms.clone());
some_or_sticky(&mut request.extensions.typing.enabled, cached.extensions.typing.enabled);
some_or_sticky(
&mut request.extensions.typing.rooms,
cached.extensions.typing.rooms.clone(),
);
some_or_sticky(
&mut request.extensions.typing.lists,
cached.extensions.typing.lists.clone(),
);
some_or_sticky(
&mut request.extensions.receipts.enabled,
cached.extensions.receipts.enabled,
);
some_or_sticky(
&mut request.extensions.receipts.rooms,
cached.extensions.receipts.rooms.clone(),
);
some_or_sticky(
&mut request.extensions.receipts.lists,
cached.extensions.receipts.lists.clone(),
);
cached.extensions = request.extensions.clone();
cached.known_rooms.clone()
}
pub fn update_sync_request_with_cache(
&self,
user_id: OwnedUserId,
@ -148,20 +293,30 @@ impl Service {
for (list_id, list) in &mut request.lists {
if let Some(cached_list) = cached.lists.get(list_id) {
list_or_sticky(&mut list.sort, &cached_list.sort);
list_or_sticky(&mut list.room_details.required_state, &cached_list.room_details.required_state);
some_or_sticky(&mut list.room_details.timeline_limit, cached_list.room_details.timeline_limit);
some_or_sticky(&mut list.include_old_rooms, cached_list.include_old_rooms.clone());
list_or_sticky(
&mut list.room_details.required_state,
&cached_list.room_details.required_state,
);
some_or_sticky(
&mut list.room_details.timeline_limit,
cached_list.room_details.timeline_limit,
);
some_or_sticky(
&mut list.include_old_rooms,
cached_list.include_old_rooms.clone(),
);
match (&mut list.filters, cached_list.filters.clone()) {
(Some(list_filters), Some(cached_filters)) => {
some_or_sticky(&mut list_filters.is_dm, cached_filters.is_dm);
list_or_sticky(&mut list_filters.spaces, &cached_filters.spaces);
some_or_sticky(&mut list_filters.is_encrypted, cached_filters.is_encrypted);
some_or_sticky(&mut list_filters.is_invite, cached_filters.is_invite);
list_or_sticky(&mut list_filters.room_types, &cached_filters.room_types);
list_or_sticky(&mut list_filters.not_room_types, &cached_filters.not_room_types);
some_or_sticky(&mut list_filters.room_name_like, cached_filters.room_name_like);
list_or_sticky(&mut list_filters.tags, &cached_filters.tags);
list_or_sticky(&mut list_filters.not_tags, &cached_filters.not_tags);
| (Some(filter), Some(cached_filter)) => {
some_or_sticky(&mut filter.is_dm, cached_filter.is_dm);
list_or_sticky(&mut filter.spaces, &cached_filter.spaces);
some_or_sticky(&mut filter.is_encrypted, cached_filter.is_encrypted);
some_or_sticky(&mut filter.is_invite, cached_filter.is_invite);
list_or_sticky(&mut filter.room_types, &cached_filter.room_types);
// Should be made possible to change
list_or_sticky(&mut filter.not_room_types, &cached_filter.not_room_types);
some_or_sticky(&mut filter.room_name_like, cached_filter.room_name_like);
list_or_sticky(&mut filter.tags, &cached_filter.tags);
list_or_sticky(&mut filter.not_tags, &cached_filter.not_tags);
},
| (_, Some(cached_filters)) => list.filters = Some(cached_filters),
| (Some(list_filters), _) => list.filters = Some(list_filters.clone()),
@ -222,18 +377,16 @@ impl Service {
subscriptions: BTreeMap<OwnedRoomId, sync_events::v4::RoomSubscription>,
) {
let mut cache = self.connections.lock().expect("locked");
let cached = Arc::clone(
cache
.entry((user_id, device_id, conn_id))
.or_insert_with(|| {
Arc::new(Mutex::new(SlidingSyncCache {
lists: BTreeMap::new(),
subscriptions: BTreeMap::new(),
known_rooms: BTreeMap::new(),
extensions: ExtensionsConfig::default(),
}))
}),
);
let cached = Arc::clone(cache.entry((user_id, device_id, conn_id)).or_insert_with(
|| {
Arc::new(Mutex::new(SlidingSyncCache {
lists: BTreeMap::new(),
subscriptions: BTreeMap::new(),
known_rooms: BTreeMap::new(),
extensions: ExtensionsConfig::default(),
}))
},
));
let cached = &mut cached.lock().expect("locked");
drop(cache);
@ -241,13 +394,18 @@ impl Service {
}
pub fn update_sync_known_rooms(
&self, user_id: OwnedUserId, device_id: OwnedDeviceId, conn_id: String, list_id: String,
new_cached_rooms: BTreeSet<OwnedRoomId>, globalsince: u64,
&self,
user_id: &UserId,
device_id: &DeviceId,
conn_id: String,
list_id: String,
new_cached_rooms: BTreeSet<OwnedRoomId>,
globalsince: u64,
) {
let mut cache = self.connections.lock().expect("locked");
let cached = Arc::clone(
cache
.entry((user_id, device_id, conn_id))
.entry((user_id.to_owned(), device_id.to_owned(), conn_id))
.or_insert_with(|| {
Arc::new(Mutex::new(SlidingSyncCache {
lists: BTreeMap::new(),
@ -275,4 +433,57 @@ impl Service {
list.insert(roomid, globalsince);
}
}
pub fn update_snake_sync_known_rooms(
&self,
user_id: &UserId,
device_id: &DeviceId,
conn_id: String,
list_id: String,
new_cached_rooms: BTreeSet<OwnedRoomId>,
globalsince: u64,
) {
let mut cache = self.snake_connections.lock().expect("locked");
let cached = Arc::clone(
cache
.entry((user_id.to_owned(), device_id.to_owned(), Some(conn_id)))
.or_insert_with(|| Arc::new(Mutex::new(SnakeSyncCache::default()))),
);
let cached = &mut cached.lock().expect("locked");
drop(cache);
for (roomid, lastsince) in cached
.known_rooms
.entry(list_id.clone())
.or_default()
.iter_mut()
{
if !new_cached_rooms.contains(roomid) {
*lastsince = 0;
}
}
let list = cached.known_rooms.entry(list_id).or_default();
for roomid in new_cached_rooms {
list.insert(roomid, globalsince);
}
}
pub fn update_snake_sync_subscriptions(
&self,
user_id: OwnedUserId,
device_id: OwnedDeviceId,
conn_id: Option<String>,
subscriptions: BTreeMap<OwnedRoomId, v5::request::RoomSubscription>,
) {
let mut cache = self.snake_connections.lock().expect("locked");
let cached = Arc::clone(
cache
.entry((user_id, device_id, conn_id))
.or_insert_with(|| Arc::new(Mutex::new(SnakeSyncCache::default()))),
);
let cached = &mut cached.lock().expect("locked");
drop(cache);
cached.subscriptions = subscriptions;
}
}