arminius_lib_sync/syncv4/execute/fetch/
rooms.rs

1// SPDX-FileCopyrightText: 2026 BWI GmbH
2//
3// SPDX-License-Identifier: Apache-2.0
4
5//! Contains logic to fetch rooms data for a sync v4 response.
6
7use std::collections::{BTreeMap, HashMap};
8
9use arminius_lib_component::error::{InternalServerError, ResultExt as _};
10use arminius_lib_db_room::EventQueries as _;
11use arminius_lib_db_sync::syncv4::{SyncedRoomTimeline, TransactedSyncV4Queries};
12use arminius_lib_events::{ContextualizedFields as _, RoomEvent, Verified};
13use itertools::{Either, Itertools as _};
14use ruma::{
15	OwnedRoomId, UInt,
16	api::client::sync::sync_events,
17	events::{AnySyncStateEvent, AnySyncTimelineEvent},
18	serde::Raw,
19};
20
21use crate::syncv4::{connect, error::SyncV4Error, plan};
22
23/// Fetches all the rooms data.
24pub(super) async fn fetch_rooms<T: TransactedSyncV4Queries>(
25	transaction: &mut T,
26	connection: &connect::ResolvedConnection,
27	rooms_bump_stamps: &HashMap<OwnedRoomId, usize>,
28	rooms_to_sync: &HashMap<OwnedRoomId, plan::RoomToSync>,
29) -> Result<BTreeMap<OwnedRoomId, sync_events::v5::response::Room>, SyncV4Error> {
30	let last_live_stream_id = connection.prev_events_stream_id().unwrap_or(0);
31	let to_events_stream_id = connection.next_token.event_stream_position;
32
33	let (joined_rooms_params, never_joined_rooms_params): (
34		HashMap<OwnedRoomId, (i64, i64)>,
35		HashMap<OwnedRoomId, bool>,
36	) = rooms_to_sync.iter().partition_map(|(room_id, room_to_sync)| match room_to_sync {
37		plan::RoomToSync::Joined { from_stream_id: sync_room_from, room_config, .. } => {
38			Either::Left((room_id.clone(), (*sync_room_from, room_config.timeline_limit)))
39		}
40		plan::RoomToSync::NeverJoined { initial } => Either::Right((room_id.to_owned(), *initial)),
41	});
42
43	let joined_rooms_timelines = transaction
44		.fetch_syncv4_joined_rooms_timelines(
45			&connection.key.user_id,
46			&joined_rooms_params,
47			last_live_stream_id,
48			to_events_stream_id,
49		)
50		.await?;
51
52	// TODO: This loop should be avoided by making
53	// `fetch_rooms_required_states()` take an HashMap directly
54	let mut joined_rooms_required_states = HashMap::new();
55
56	// Fetch all the required states for joined rooms.
57	for (room_id, from_stream_id, required_state) in
58		rooms_to_sync.iter().filter_map(|(room_id, room_to_sync)| match room_to_sync {
59			plan::RoomToSync::Joined { from_stream_id, room_config, .. } => {
60				Some((room_id, from_stream_id, &room_config.required_state))
61			}
62			_ => None,
63		}) {
64		let required_states = transaction
65			.fetch_rooms_required_states(
66				room_id,
67				required_state,
68				// TODO: Make this non-optional ?
69				*from_stream_id,
70				to_events_stream_id,
71			)
72			.await?;
73		joined_rooms_required_states.insert(room_id.clone(), required_states);
74	}
75
76	let mut rooms = compute_joined_rooms(
77		transaction,
78		connection,
79		joined_rooms_timelines,
80		joined_rooms_required_states,
81		rooms_bump_stamps,
82		rooms_to_sync,
83	)
84	.await?;
85
86	let mut never_joined_rooms =
87		compute_never_joined_rooms(transaction, never_joined_rooms_params, rooms_bump_stamps)
88			.await?;
89
90	rooms.append(&mut never_joined_rooms);
91
92	Ok(rooms)
93}
94
95/// Given the timelines of rooms the user is or has been a member of,
96/// contextualize each event, and fetch the various other fields needed to
97/// compute a [`sync_events::v5::response::Room`].
98async fn compute_joined_rooms<T: TransactedSyncV4Queries>(
99	transaction: &mut T,
100	connection: &connect::ResolvedConnection,
101	rooms_timelines: HashMap<OwnedRoomId, SyncedRoomTimeline>,
102	rooms_required_states: HashMap<OwnedRoomId, Vec<RoomEvent<Verified>>>,
103	rooms_bump_stamps: &HashMap<OwnedRoomId, usize>,
104	rooms_to_sync: &HashMap<OwnedRoomId, plan::RoomToSync>,
105) -> Result<BTreeMap<OwnedRoomId, sync_events::v5::response::Room>, SyncV4Error> {
106	let mut result = BTreeMap::new();
107
108	for (room_id, room_timeline) in rooms_timelines {
109		let mut room = sync_events::v5::response::Room::new();
110
111		room.initial = Some(
112			rooms_to_sync
113				.get(&room_id)
114				.and_then(|room_to_sync| match room_to_sync {
115					plan::RoomToSync::Joined { initial, .. } => Some(*initial),
116					plan::RoomToSync::NeverJoined { .. } => None,
117				})
118				.unrecoverable_error(
119					"Fetched a timeline for a room that is not a joined room of the plan",
120				)?,
121		);
122
123		room.bump_stamp = Some(
124			UInt::try_from(
125				*rooms_bump_stamps.get(&room_id).unrecoverable_error("Missing room bump stamp")?,
126			)
127			.unrecoverable_error("Failed to create UInt")?,
128		);
129
130		// PERF: This might be costly and hence have its place in some
131		// pre-populated `syncv4_rooms_info` table.
132		let (joined_count, invited_count) =
133			transaction.fetch_room_membership_counts(&room_id).await?;
134
135		room.joined_count =
136			Some(UInt::try_from(joined_count).unrecoverable_error("Failed to create UInt")?);
137		room.invited_count =
138			Some(UInt::try_from(invited_count).unrecoverable_error("Failed to create UInt")?);
139
140		// Whether the `timeline_limit` was hit
141		room.limited = room_timeline.limited;
142
143		// Number of events that occurred after the previous sync
144		room.num_live = Some(
145			UInt::try_from(room_timeline.num_live).unrecoverable_error("Failed to create UInt")?,
146		);
147
148		// Contextualize each event of the timeline for this specific user's
149		// device
150		for event in room_timeline.timeline {
151			let client_context = transaction
152				.fetch_client_context(
153					event.room_id(),
154					event.event_id(),
155					&connection.key.user_id,
156					&connection.key.device_id,
157				)
158				.await?;
159
160			room.timeline.push(Raw::<AnySyncTimelineEvent>::from_json(
161				event.to_client_event(client_context),
162			));
163		}
164
165		result.insert(room_id, room);
166	}
167
168	// Also contextualize required states events
169	for (room_id, room_required_state) in rooms_required_states {
170		for event in room_required_state {
171			let client_context = transaction
172				.fetch_client_context(
173					event.room_id(),
174					event.event_id(),
175					&connection.key.user_id,
176					&connection.key.device_id,
177				)
178				.await?;
179
180			let room = result.get_mut(&room_id).unrecoverable_error(
181				"Fetching required states for a room without fetching its timeline ?",
182			)?;
183
184			room.required_state
185				.push(Raw::<AnySyncStateEvent>::from_json(event.to_client_event(client_context)));
186		}
187	}
188
189	Ok(result)
190}
191
192/// Fetches all the data for the rooms the user never joined.
193async fn compute_never_joined_rooms<T: TransactedSyncV4Queries>(
194	transaction: &mut T,
195	never_joined_rooms: HashMap<OwnedRoomId, bool>,
196	rooms_bump_stamps: &HashMap<OwnedRoomId, usize>,
197) -> Result<BTreeMap<OwnedRoomId, sync_events::v5::response::Room>, InternalServerError> {
198	let never_joined_room_ids: Vec<OwnedRoomId> = never_joined_rooms.keys().cloned().collect();
199
200	transaction
201		.fetch_rooms_stripped_state(&never_joined_room_ids)
202		.await?
203		.into_iter()
204		.map(|(room_id, invite_state)| {
205			let mut room = sync_events::v5::response::Room::new();
206
207			room.bump_stamp = Some(
208				UInt::try_from(
209					*rooms_bump_stamps
210						.get(&room_id)
211						.unrecoverable_error("Missing room bump stamp")?,
212				)
213				.unrecoverable_error("Failed to create UInt")?,
214			);
215			room.invite_state = Some(invite_state);
216			room.initial = Some(*never_joined_rooms.get(&room_id).unrecoverable_error(
217				"Fetched a stripped state for a room that was not in the plan",
218			)?);
219
220			Ok::<_, InternalServerError>((room_id, room))
221		})
222		.collect::<Result<BTreeMap<_, _>, _>>()
223}
arminius_lib_sync/syncv4/execute/fetch/rooms.rs

arminius_lib_sync/syncv4/execute/fetch/
rooms.rs
