oss/conduit
oss/conduit
1
0
Fork 0
mirror of https://gitlab.com/famedly/conduit.git synced 2025-06-27 16:35:59 +00:00
Code Wiki Activity

feat(sync): v4 -> v5

Browse source
This commit is contained in:
Matthias Ahouansou 2025-05-07 21:36:57 +01:00
parent 215198d1c0
commit ec2c61e277
No known key found for this signature in database
6 changed files with 66 additions and 128 deletions
Download patch file Download diff file Expand all files Collapse all files

2
Cargo.toml
View file

@ -171,7 +171,7 @@ features = [
"ring-compat",
"state-res",
"unstable-msc2448",
"unstable-msc3575",
"unstable-msc4186",
]
git = "https://github.com/ruma/ruma.git"

128
src/api/client_server/sync.rs
View file

@ -13,7 +13,6 @@ use ruma::{
KnockState, KnockedRoom, LeftRoom, Presence, RoomAccountData, RoomSummary, Rooms,
State, Timeline, ToDevice,
},
v4::{SlidingOp, SlidingSyncRoomHero},
DeviceLists, UnreadNotificationsCount,
},
uiaa::UiaaResponse,
@ -1270,9 +1269,9 @@ fn share_encrypted_room(
.any(|encrypted| encrypted))
}
pub async fn sync_events_v4_route(
body: Ruma<sync_events::v4::Request>,
) -> Result<sync_events::v4::Response, RumaResponse<UiaaResponse>> {
pub async fn sync_events_v5_route(
body: Ruma<sync_events::v5::Request>,
) -> Result<sync_events::v5::Response, RumaResponse<UiaaResponse>> {
let sender_user = body.sender_user.expect("user is authenticated");
let sender_device = body.sender_device.expect("user is authenticated");
let mut body = body.body;
@ -1331,13 +1330,12 @@ pub async fn sync_events_v4_route(
);
for room_id in &all_joined_rooms {
let current_shortstatehash =
if let Some(s) = services().rooms.state.get_room_shortstatehash(room_id)? {
s
} else {
error!("Room {} has no state", room_id);
continue;
};
let Some(current_shortstatehash) =
services().rooms.state.get_room_shortstatehash(room_id)?
else {
error!("Room {} has no state", room_id);
continue;
};
let since_shortstatehash = services()
.rooms
@ -1510,54 +1508,36 @@ pub async fn sync_events_v4_route(
let mut new_known_rooms = BTreeSet::new();
for (mut start, mut end) in list.ranges {
start = start.clamp(uint!(0), UInt::from(all_joined_rooms.len() as u32 - 1));
end = end.clamp(start, UInt::from(all_joined_rooms.len() as u32 - 1));
let room_ids =
all_joined_rooms[(u64::from(start) as usize)..=(u64::from(end) as usize)].to_vec();
new_known_rooms.extend(room_ids.iter().cloned());
for room_id in &room_ids {
let todo_room =
todo_rooms
.entry(room_id.clone())
.or_insert((BTreeSet::new(), 0, u64::MAX));
let limit = u64::from(list.room_details.timeline_limit).min(100);
todo_room
.0
.extend(list.room_details.required_state.iter().cloned());
todo_room.1 = todo_room.1.max(limit);
// 0 means unknown because it got out of date
todo_room.2 = todo_room.2.min(
known_rooms
.get(&list_id)
.and_then(|k| k.get(room_id))
.copied()
.unwrap_or(0),
);
}
}
lists.insert(
list_id.clone(),
sync_events::v4::SyncList {
ops: list
.ranges
.into_iter()
.map(|mut r| {
r.0 =
r.0.clamp(uint!(0), UInt::from(all_joined_rooms.len() as u32 - 1));
r.1 =
r.1.clamp(r.0, UInt::from(all_joined_rooms.len() as u32 - 1));
let room_ids = all_joined_rooms
[(u64::from(r.0) as usize)..=(u64::from(r.1) as usize)]
.to_vec();
new_known_rooms.extend(room_ids.iter().cloned());
for room_id in &room_ids {
let todo_room = todo_rooms.entry(room_id.clone()).or_insert((
BTreeSet::new(),
0,
u64::MAX,
));
let limit = list
.room_details
.timeline_limit
.map_or(10, u64::from)
.min(100);
todo_room
.0
.extend(list.room_details.required_state.iter().cloned());
todo_room.1 = todo_room.1.max(limit);
// 0 means unknown because it got out of date
todo_room.2 = todo_room.2.min(
known_rooms
.get(&list_id)
.and_then(|k| k.get(room_id))
.copied()
.unwrap_or(0),
);
}
sync_events::v4::SyncOp {
op: SlidingOp::Sync,
range: Some(r),
index: None,
room_ids,
room_id: None,
}
})
.collect(),
sync_events::v5::response::List {
count: UInt::from(all_joined_rooms.len() as u32),
},
);
@ -1582,7 +1562,7 @@ pub async fn sync_events_v4_route(
let todo_room = todo_rooms
.entry(room_id.clone())
.or_insert((BTreeSet::new(), 0, u64::MAX));
let limit = room.timeline_limit.map_or(10, u64::from).min(100);
let limit = u64::from(room.timeline_limit).min(100);
todo_room.0.extend(room.required_state.iter().cloned());
todo_room.1 = todo_room.1.max(limit);
// 0 means unknown because it got out of date
@ -1596,11 +1576,6 @@ pub async fn sync_events_v4_route(
known_subscription_rooms.insert(room_id.clone());
}
for r in body.unsubscribe_rooms {
known_subscription_rooms.remove(&r);
body.room_subscriptions.remove(&r);
}
if let Some(conn_id) = &body.conn_id {
services().users.update_sync_known_rooms(
sender_user.clone(),
@ -1656,6 +1631,11 @@ pub async fn sync_events_v4_route(
.map(|(_, pdu)| pdu.to_sync_room_event())
.collect();
let bump_stamp = timeline_pdus
.iter()
.map(|(_, pdu)| pdu.origin_server_ts)
.max();
let required_state = required_state_request
.iter()
.flat_map(|state| {
@ -1683,7 +1663,7 @@ pub async fn sync_events_v4_route(
.get_member(room_id, &member)
.ok()
.flatten()
.map(|memberevent| SlidingSyncRoomHero {
.map(|memberevent| sync_events::v5::response::Hero {
user_id: member,
name: memberevent.displayname,
avatar: memberevent.avatar_url,
@ -1720,7 +1700,7 @@ pub async fn sync_events_v4_route(
rooms.insert(
room_id.clone(),
sync_events::v4::SlidingSyncRoom {
sync_events::v5::response::Room {
name: services().rooms.state_accessor.get_name(room_id)?.or(name),
avatar: if let Some(avatar) = avatar {
JsOption::Some(avatar)
@ -1773,7 +1753,7 @@ pub async fn sync_events_v4_route(
.into(),
),
num_live: None, // Count events in timeline greater than global sync counter
timestamp: None,
bump_stamp,
heroes: if body
.room_subscriptions
.get(room_id)
@ -1801,15 +1781,14 @@ pub async fn sync_events_v4_route(
let _ = tokio::time::timeout(duration, watcher).await;
}
Ok(sync_events::v4::Response {
initial: globalsince == 0,
Ok(sync_events::v5::Response {
txn_id: body.txn_id.clone(),
pos: next_batch.to_string(),
lists,
rooms,
extensions: sync_events::v4::Extensions {
extensions: sync_events::v5::response::Extensions {
to_device: if body.extensions.to_device.enabled.unwrap_or(false) {
Some(sync_events::v4::ToDevice {
Some(sync_events::v5::response::ToDevice {
events: services()
.users
.get_to_device_events(&sender_user, &sender_device)?,
@ -1818,7 +1797,7 @@ pub async fn sync_events_v4_route(
} else {
None
},
e2ee: sync_events::v4::E2EE {
e2ee: sync_events::v5::response::E2EE {
device_lists: DeviceLists {
changed: device_list_changes.into_iter().collect(),
left: device_list_left.into_iter().collect(),
@ -1829,7 +1808,7 @@ pub async fn sync_events_v4_route(
// Fallback keys are not yet supported
device_unused_fallback_key_types: None,
},
account_data: sync_events::v4::AccountData {
account_data: sync_events::v5::response::AccountData {
global: if body.extensions.account_data.enabled.unwrap_or(false) {
services()
.account_data
@ -1848,13 +1827,12 @@ pub async fn sync_events_v4_route(
},
rooms: BTreeMap::new(),
},
receipts: sync_events::v4::Receipts {
receipts: sync_events::v5::response::Receipts {
rooms: BTreeMap::new(),
},
typing: sync_events::v4::Typing {
typing: sync_events::v5::response::Typing {
rooms: BTreeMap::new(),
},
},
delta_token: None,
})
}

1
src/api/client_server/unversioned.rs
View file

@ -30,6 +30,7 @@ pub async fn get_supported_versions_route(
unstable_features: BTreeMap::from_iter([
("org.matrix.e2e_cross_signing".to_owned(), true),
("org.matrix.msc3916.stable".to_owned(), true),
("org.matrix.simplified_msc3575".to_owned(), true),
]),
};

5
src/api/client_server/well_known.rs
View file

@ -1,6 +1,4 @@
use ruma::api::client::discovery::discover_homeserver::{
self, HomeserverInfo, SlidingSyncProxyInfo,
};
use ruma::api::client::discovery::discover_homeserver::{self, HomeserverInfo};
use crate::{services, Result, Ruma};
@ -17,6 +15,5 @@ pub async fn well_known_client(
base_url: client_url.clone(),
},
identity_server: None,
sliding_sync_proxy: Some(SlidingSyncProxyInfo { url: client_url }),
})
}

2
src/main.rs
View file

@ -410,7 +410,7 @@ fn routes(config: &Config) -> Router {
.put(client_server::send_state_event_for_empty_key_route),
)
.ruma_route(client_server::sync_events_route)
.ruma_route(client_server::sync_events_v4_route)
.ruma_route(client_server::sync_events_v5_route)
.ruma_route(client_server::get_context_route)
.ruma_route(client_server::get_message_events_route)
.ruma_route(client_server::search_events_route)

56
src/service/users/mod.rs
View file

@ -10,10 +10,7 @@ use ruma::{
api::client::{
device::Device,
filter::FilterDefinition,
sync::sync_events::{
self,
v4::{ExtensionsConfig, SyncRequestList},
},
sync::sync_events::{self},
},
encryption::{CrossSigningKey, DeviceKeys, OneTimeKey},
events::AnyToDeviceEvent,
@ -25,10 +22,10 @@ use ruma::{
use crate::{services, Error, Result};
pub struct SlidingSyncCache {
lists: BTreeMap<String, SyncRequestList>,
subscriptions: BTreeMap<OwnedRoomId, sync_events::v4::RoomSubscription>,
lists: BTreeMap<String, sync_events::v5::request::List>,
subscriptions: BTreeMap<OwnedRoomId, sync_events::v5::request::RoomSubscription>,
known_rooms: BTreeMap<String, BTreeMap<OwnedRoomId, u64>>, // For every room, the roomsince number
extensions: ExtensionsConfig,
extensions: sync_events::v5::request::Extensions,
}
pub struct Service {
@ -60,7 +57,7 @@ impl Service {
&self,
user_id: OwnedUserId,
device_id: OwnedDeviceId,
request: &mut sync_events::v4::Request,
request: &mut sync_events::v5::Request,
) -> BTreeMap<String, BTreeMap<OwnedRoomId, u64>> {
let Some(conn_id) = request.conn_id.clone() else {
return BTreeMap::new();
@ -75,7 +72,7 @@ impl Service {
lists: BTreeMap::new(),
subscriptions: BTreeMap::new(),
known_rooms: BTreeMap::new(),
extensions: ExtensionsConfig::default(),
extensions: sync_events::v5::request::Extensions::default(),
}))
}),
);
@ -84,57 +81,22 @@ impl Service {
for (list_id, list) in &mut request.lists {
if let Some(cached_list) = cached.lists.get(list_id) {
if list.sort.is_empty() {
list.sort.clone_from(&cached_list.sort);
};
if list.room_details.required_state.is_empty() {
list.room_details
.required_state
.clone_from(&cached_list.room_details.required_state);
};
list.room_details.timeline_limit = list
.room_details
.timeline_limit
.or(cached_list.room_details.timeline_limit);
list.include_old_rooms = list
.include_old_rooms
.clone()
.or(cached_list.include_old_rooms.clone());
match (&mut list.filters, cached_list.filters.clone()) {
(Some(list_filters), Some(cached_filters)) => {
list_filters.is_dm = list_filters.is_dm.or(cached_filters.is_dm);
if list_filters.spaces.is_empty() {
list_filters.spaces = cached_filters.spaces;
}
list_filters.is_encrypted =
list_filters.is_encrypted.or(cached_filters.is_encrypted);
list_filters.is_invite =
list_filters.is_invite.or(cached_filters.is_invite);
if list_filters.room_types.is_empty() {
list_filters.room_types = cached_filters.room_types;
}
if list_filters.not_room_types.is_empty() {
list_filters.not_room_types = cached_filters.not_room_types;
}
list_filters.room_name_like = list_filters
.room_name_like
.clone()
.or(cached_filters.room_name_like);
if list_filters.tags.is_empty() {
list_filters.tags = cached_filters.tags;
}
if list_filters.not_tags.is_empty() {
list_filters.not_tags = cached_filters.not_tags;
}
}
(_, Some(cached_filters)) => list.filters = Some(cached_filters),
(Some(list_filters), _) => list.filters = Some(list_filters.clone()),
(_, _) => {}
}
if list.bump_event_types.is_empty() {
list.bump_event_types
.clone_from(&cached_list.bump_event_types);
};
}
cached.lists.insert(list_id.clone(), list.clone());
}
@ -192,7 +154,7 @@ impl Service {
user_id: OwnedUserId,
device_id: OwnedDeviceId,
conn_id: String,
subscriptions: BTreeMap<OwnedRoomId, sync_events::v4::RoomSubscription>,
subscriptions: BTreeMap<OwnedRoomId, sync_events::v5::request::RoomSubscription>,
) {
let mut cache = self.connections.lock().unwrap();
let cached = Arc::clone(
@ -203,7 +165,7 @@ impl Service {
lists: BTreeMap::new(),
subscriptions: BTreeMap::new(),
known_rooms: BTreeMap::new(),
extensions: ExtensionsConfig::default(),
extensions: sync_events::v5::request::Extensions::default(),
}))
}),
);
@ -231,7 +193,7 @@ impl Service {
lists: BTreeMap::new(),
subscriptions: BTreeMap::new(),
known_rooms: BTreeMap::new(),
extensions: ExtensionsConfig::default(),
extensions: sync_events::v5::request::Extensions::default(),
}))
}),
);