1use std::{
18 collections::{BTreeMap, btree_map},
19 fmt,
20 time::Duration,
21};
22
23pub use matrix_sdk_base::sync::*;
24use matrix_sdk_base::{
25 debug::{
26 DebugInvitedRoom, DebugKnockedRoom, DebugListOfProcessedToDeviceEvents,
27 DebugListOfRawEventsNoId,
28 },
29 sleep::sleep,
30 sync::SyncResponse as BaseSyncResponse,
31 timer,
32};
33use matrix_sdk_common::deserialized_responses::ProcessedToDeviceEvent;
34use ruma::{
35 OwnedRoomId, RoomId,
36 api::client::sync::sync_events::{
37 self,
38 v3::{InvitedRoom, KnockedRoom},
39 },
40 events::{AnyGlobalAccountDataEvent, presence::PresenceEvent},
41 serde::Raw,
42 time::Instant,
43};
44use tracing::{debug, error, warn};
45
46use crate::{Client, Result, Room, event_handler::HandlerKind};
47
48#[derive(Clone, Default)]
50pub struct SyncResponse {
51 pub next_batch: String,
54 pub rooms: RoomUpdates,
56 pub presence: Vec<Raw<PresenceEvent>>,
58 pub account_data: Vec<Raw<AnyGlobalAccountDataEvent>>,
60 pub to_device: Vec<ProcessedToDeviceEvent>,
62 pub notifications: BTreeMap<OwnedRoomId, Vec<Notification>>,
64}
65
66impl SyncResponse {
67 pub(crate) fn new(next_batch: String, base_response: BaseSyncResponse) -> Self {
68 let BaseSyncResponse { rooms, presence, account_data, to_device, notifications } =
69 base_response;
70
71 Self { next_batch, rooms, presence, account_data, to_device, notifications }
72 }
73}
74
75#[cfg(not(tarpaulin_include))]
76impl fmt::Debug for SyncResponse {
77 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
78 f.debug_struct("SyncResponse")
79 .field("next_batch", &self.next_batch)
80 .field("rooms", &self.rooms)
81 .field("account_data", &DebugListOfRawEventsNoId(&self.account_data))
82 .field("to_device", &DebugListOfProcessedToDeviceEvents(&self.to_device))
83 .field("notifications", &self.notifications)
84 .finish_non_exhaustive()
85 }
86}
87
88#[derive(Clone)]
90pub enum RoomUpdate {
91 Left {
93 room: Room,
95 updates: LeftRoomUpdate,
97 },
98 Joined {
100 room: Room,
102 updates: JoinedRoomUpdate,
104 },
105 Invited {
107 room: Room,
109 updates: InvitedRoom,
111 },
112 Knocked {
114 room: Room,
116 updates: KnockedRoom,
118 },
119}
120
121#[cfg(not(tarpaulin_include))]
122impl fmt::Debug for RoomUpdate {
123 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
124 match self {
125 Self::Left { room, updates } => {
126 f.debug_struct("Left").field("room", room).field("updates", updates).finish()
127 }
128 Self::Joined { room, updates } => {
129 f.debug_struct("Joined").field("room", room).field("updates", updates).finish()
130 }
131 Self::Invited { room, updates } => f
132 .debug_struct("Invited")
133 .field("room", room)
134 .field("updates", &DebugInvitedRoom(updates))
135 .finish(),
136 Self::Knocked { room, updates } => f
137 .debug_struct("Knocked")
138 .field("room", room)
139 .field("updates", &DebugKnockedRoom(updates))
140 .finish(),
141 }
142 }
143}
144
145impl Client {
148 pub(crate) async fn process_sync(
151 &self,
152 response: sync_events::v3::Response,
153 ) -> Result<BaseSyncResponse> {
154 let response = Box::pin(self.base_client().receive_sync_response(response)).await?;
155
156 #[cfg(feature = "e2e-encryption")]
158 self.encryption().backups().maybe_trigger_backup();
159
160 self.call_sync_response_handlers(&response).await?;
161
162 Ok(response)
163 }
164
165 #[tracing::instrument(skip(self, response))]
172 pub(crate) async fn call_sync_response_handlers(
173 &self,
174 response: &BaseSyncResponse,
175 ) -> Result<()> {
176 let _timer = timer!(tracing::Level::TRACE, "_method");
177
178 let BaseSyncResponse { rooms, presence, account_data, to_device, notifications } = response;
179
180 let now = Instant::now();
181 self.handle_sync_events(HandlerKind::GlobalAccountData, None, account_data).await?;
182 self.handle_sync_events(HandlerKind::Presence, None, presence).await?;
183 self.handle_sync_to_device_events(to_device).await?;
184
185 let _ = self.inner.room_updates_sender.send(rooms.clone());
187
188 for (room_id, room_info) in &rooms.joined {
189 let Some(room) = self.get_room(room_id) else {
190 error!(?room_id, "Can't call event handler, room not found");
191 continue;
192 };
193
194 self.send_room_update(room_id, || RoomUpdate::Joined {
195 room: room.clone(),
196 updates: room_info.clone(),
197 });
198
199 let JoinedRoomUpdate {
200 unread_notifications: _,
201 timeline,
202 state,
203 account_data,
204 ephemeral,
205 ambiguity_changes: _,
206 } = room_info;
207
208 let room = Some(&room);
209 self.handle_sync_events(HandlerKind::RoomAccountData, room, account_data).await?;
210 self.handle_sync_state_events(room, state).await?;
211 self.handle_sync_timeline_events(room, &timeline.events).await?;
212 self.handle_sync_events(HandlerKind::EphemeralRoomData, room, ephemeral).await?;
215 }
216
217 for (room_id, room_info) in &rooms.left {
218 let Some(room) = self.get_room(room_id) else {
219 error!(?room_id, "Can't call event handler, room not found");
220 continue;
221 };
222
223 self.send_room_update(room_id, || RoomUpdate::Left {
224 room: room.clone(),
225 updates: room_info.clone(),
226 });
227
228 let LeftRoomUpdate { timeline, state, account_data, ambiguity_changes: _ } = room_info;
229
230 let room = Some(&room);
231 self.handle_sync_events(HandlerKind::RoomAccountData, room, account_data).await?;
232 self.handle_sync_state_events(room, state).await?;
233 self.handle_sync_timeline_events(room, &timeline.events).await?;
234 }
235
236 for (room_id, room_info) in &rooms.invited {
237 let Some(room) = self.get_room(room_id) else {
238 error!(?room_id, "Can't call event handler, room not found");
239 continue;
240 };
241
242 self.send_room_update(room_id, || RoomUpdate::Invited {
243 room: room.clone(),
244 updates: room_info.clone(),
245 });
246
247 let invite_state = &room_info.invite_state.events;
248 self.handle_sync_events(HandlerKind::StrippedState, Some(&room), invite_state).await?;
249 }
250
251 for (room_id, room_info) in &rooms.knocked {
252 let Some(room) = self.get_room(room_id) else {
253 error!(?room_id, "Can't call event handler, room not found");
254 continue;
255 };
256
257 self.send_room_update(room_id, || RoomUpdate::Knocked {
258 room: room.clone(),
259 updates: room_info.clone(),
260 });
261
262 let knock_state = &room_info.knock_state.events;
263 self.handle_sync_events(HandlerKind::StrippedState, Some(&room), knock_state).await?;
264 }
265
266 debug!("Ran event handlers in {:?}", now.elapsed());
267
268 let now = Instant::now();
269
270 let mut futures = Vec::new();
272 for handler in &*self.notification_handlers().await {
273 for (room_id, room_notifications) in notifications {
274 let Some(room) = self.get_room(room_id) else {
275 warn!(?room_id, "Can't call notification handler, room not found");
276 continue;
277 };
278
279 futures.extend(room_notifications.iter().map(|notification| {
280 (handler)(notification.clone(), room.clone(), self.clone())
281 }));
282 }
283 }
284
285 for fut in futures {
288 fut.await;
289 }
290
291 debug!("Ran notification handlers in {:?}", now.elapsed());
292
293 Ok(())
294 }
295
296 fn send_room_update(&self, room_id: &RoomId, make_msg: impl FnOnce() -> RoomUpdate) {
297 if let btree_map::Entry::Occupied(entry) =
298 self.inner.room_update_channels.lock().unwrap().entry(room_id.to_owned())
299 {
300 let tx = entry.get();
301 if tx.receiver_count() == 0 {
302 entry.remove();
303 } else {
304 _ = tx.send(make_msg());
305 }
306 }
307 }
308
309 async fn sleep() {
310 sleep(Duration::from_secs(1)).await;
311 }
312
313 pub(crate) async fn sync_loop_helper(
314 &self,
315 sync_settings: &mut crate::config::SyncSettings,
316 ) -> Result<SyncResponse> {
317 let response = self.sync_once(sync_settings.clone()).await;
318
319 match response {
320 Ok(r) => {
321 sync_settings.token = r.next_batch.clone().into();
322 Ok(r)
323 }
324 Err(e) => {
325 error!("Received an invalid response: {e}");
326 Err(e)
327 }
328 }
329 }
330
331 pub(crate) async fn delay_sync(last_sync_time: &mut Option<Instant>) {
332 let now = Instant::now();
333
334 if let Some(t) = last_sync_time
338 && now - *t <= Duration::from_secs(1)
339 {
340 Self::sleep().await;
341 }
342
343 *last_sync_time = Some(now);
344 }
345}