tuwunel_api/client/room/
initial_sync.rs

1use axum::extract::State;
2use futures::{
3	FutureExt, TryFutureExt, TryStreamExt,
4	future::{ok, try_join, try_join4},
5};
6use ruma::{
7	RoomId,
8	api::client::room::initial_sync::v3::{PaginationChunk, Request, Response},
9	events::{
10		AnyRawAccountDataEvent,
11		StateEventType::RoomMember,
12		room::member::{MembershipState, RoomMemberEventContent},
13	},
14};
15use tuwunel_core::{
16	Event, Result, at, err, extract_variant,
17	matrix::{Pdu, PduCount},
18	utils::{
19		BoolExt, TryReadyExt,
20		result::NotFound,
21		stream::{TryTools, TryWidebandExt},
22	},
23};
24use tuwunel_service::rooms::short::ShortStateHash;
25
26use crate::{Ruma, client::visibility_filter};
27
28const LIMIT_MAX: usize = 50;
29
30/// GET `/_matrix/client/v3/rooms/{roomId}/initialSync`
31pub(crate) async fn room_initial_sync_route(
32	State(services): State<crate::State>,
33	body: Ruma<Request>,
34) -> Result<Response> {
35	let room_id = &body.room_id;
36	let sender_user = body.sender_user();
37
38	// `user_membership` uses `Ban` when a once-joined user's left row was forgotten.
39	let cached_membership = services
40		.state_cache
41		.user_membership(sender_user, room_id)
42		.await;
43
44	matches!(cached_membership.as_ref(), Some(MembershipState::Ban))
45		.is_false()
46		.ok_or_else(|| err!(Request(Forbidden("No room preview available."))))?;
47
48	services
49		.state_accessor
50		.user_can_see_state_events(sender_user, room_id)
51		.await
52		.ok_or_else(|| err!(Request(Forbidden("No room preview available."))))?;
53
54	let current_shortstatehash = services
55		.state
56		.get_room_shortstatehash(room_id)
57		.await?;
58
59	let member = services
60		.state_accessor
61		.state_get(current_shortstatehash, &RoomMember, sender_user.as_str())
62		.await
63		.optional()?;
64
65	let membership = member
66		.as_ref()
67		.map(Event::get_content)
68		.transpose()?
69		.map(|content: RoomMemberEventContent| content.membership);
70
71	let next_batch = services.globals.current_count();
72	let departure = membership
73		.as_ref()
74		.filter(|membership| matches!(membership, MembershipState::Leave | MembershipState::Ban))
75		.zip(member.as_ref())
76		.map(|(membership, pdu)| {
77			departure_snapshot(
78				&services,
79				room_id,
80				pdu,
81				membership.to_owned(),
82				current_shortstatehash,
83			)
84			.left_future()
85		});
86
87	let current_snapshot = ok((PduCount::Normal(next_batch), current_shortstatehash, membership));
88	let (timeline_end, shortstatehash, membership) = departure
89		.unwrap_or_else(|| current_snapshot.right_future())
90		.await?;
91
92	let visibility = services.directory.visibility(room_id).map(Ok);
93	let limit = body.limit.unwrap_or(LIMIT_MAX).min(LIMIT_MAX);
94	let state = services
95		.state_accessor
96		.state_full_pdus_strict(shortstatehash)
97		.map_ok(Event::into_format)
98		.try_collect::<Vec<_>>();
99
100	let events = services
101		.timeline
102		.pdus_rev(Some(sender_user), room_id, Some(timeline_end.saturating_add(1)))
103		.wide_and_then(|item| visibility_filter(&services, item, sender_user).map(Ok))
104		.ready_try_filter_map(Ok)
105		.try_take(limit)
106		.try_collect()
107		.map_ok(|mut vec: Vec<_>| {
108			vec.reverse();
109			vec
110		});
111
112	let account_data = services
113		.account_data
114		.changes_since_fallible(
115			Some(room_id),
116			sender_user,
117			0,
118			Some(timeline_end.into_normal().into_unsigned()),
119		)
120		.ready_try_filter_map(|e| Ok(extract_variant!(e, AnyRawAccountDataEvent::Room)))
121		.try_collect::<Vec<_>>();
122
123	let (visibility, state, events, account_data) = try_join4(visibility, state, events, account_data)
124			.boxed() // erase the state stream's higher-ranked event lifetime
125			.await?;
126
127	Ok(Response {
128		room_id: room_id.to_owned(),
129		membership,
130		visibility: visibility.into(),
131		account_data: Some(account_data),
132		state: state.into(),
133		messages: PaginationChunk {
134			start: events
135				.first()
136				.map(at!(0))
137				.as_ref()
138				.map(ToString::to_string),
139
140			end: events
141				.last()
142				.map(at!(0))
143				.as_ref()
144				.map_or_else(|| timeline_end.to_string(), ToString::to_string),
145
146			chunk: events
147				.into_iter()
148				.map(at!(1))
149				.map(Event::into_format)
150				.collect(),
151		}
152		.into(),
153	})
154}
155
156async fn departure_snapshot(
157	services: &crate::State,
158	room_id: &RoomId,
159	pdu: &Pdu,
160	membership: MembershipState,
161	current_shortstatehash: ShortStateHash,
162) -> Result<(PduCount, ShortStateHash, Option<MembershipState>)> {
163	let timeline_end = services.timeline.get_pdu_count(pdu.event_id());
164	let latest_count = services
165		.timeline
166		.last_timeline_count(None, room_id, None);
167
168	let (timeline_end, latest_count) = try_join(timeline_end, latest_count).await?;
169
170	let shortstatehash = (latest_count == timeline_end)
171		.then(|| ok(current_shortstatehash).left_future())
172		.unwrap_or_else(|| {
173			services
174				.timeline
175				.next_shortstatehash(room_id, timeline_end)
176				.right_future()
177		})
178		.await?;
179
180	Ok((timeline_end, shortstatehash, Some(membership)))
181}
tuwunel_api/client/room/initial_sync.rs

tuwunel_api/client/room/
initial_sync.rs
