tuwunel_api_client/room/
initial_sync.rs

1use axum::extract::State;
2use futures::{FutureExt, TryFutureExt, TryStreamExt, future::try_join4};
3use ruma::{
4	api::client::room::initial_sync::v3::{PaginationChunk, Request, Response},
5	events::{
6		AnyRawAccountDataEvent, StateEventType::RoomMember, room::member::RoomMemberEventContent,
7	},
8};
9use tuwunel_core::{
10	Result, at, err, extract_variant,
11	utils::{
12		BoolExt, TryReadyExt,
13		result::NotFound,
14		stream::{TryTools, TryWidebandExt},
15	},
16};
17use tuwunel_matrix::{Event, PduCount};
18
19use crate::{Ruma, visibility_filter};
20
21const LIMIT_MAX: usize = 50;
22
23/// GET `/_matrix/client/v3/rooms/{roomId}/initialSync`
24pub(crate) async fn room_initial_sync_route(
25	State(services): State<crate::State>,
26	body: Ruma<Request>,
27) -> Result<Response> {
28	let room_id = &body.room_id;
29	let sender_user = body.sender_user();
30
31	services
32		.state_accessor
33		.user_can_see_state_events(sender_user, room_id)
34		.await
35		.into_option()
36		.ok_or_else(|| err!(Request(Forbidden("No room preview available."))))?;
37
38	let current_shortstatehash = services
39		.state
40		.get_room_shortstatehash(room_id)
41		.await?;
42
43	let membership = services
44		.state_accessor
45		.state_get(current_shortstatehash, &RoomMember, sender_user.as_str())
46		.await
47		.optional()?
48		.as_ref()
49		.map(Event::get_content)
50		.transpose()?
51		.map(|content: RoomMemberEventContent| content.membership);
52
53	let departure = services
54		.state_accessor
55		.departure_count(sender_user, room_id)
56		.await?;
57
58	let next_batch = services.globals.current_count();
59	let timeline_end = departure.unwrap_or(PduCount::Normal(next_batch));
60	let shortstatehash = services
61		.state_accessor
62		.user_visible_shortstatehash(sender_user, room_id, None)
63		.map(|result| {
64			result?.ok_or_else(|| err!(Request(Forbidden("No room preview available."))))
65		})
66		.await?;
67
68	let visibility = services.directory.visibility(room_id).map(Ok);
69	let limit = body.limit.unwrap_or(LIMIT_MAX).min(LIMIT_MAX);
70	let state = services
71		.state_accessor
72		.state_full_pdus_strict(shortstatehash)
73		.map_ok(Event::into_pdu)
74		.map_ok(|pdu| pdu.without_private_unless_sender(Some(sender_user)))
75		.map_ok(Event::into_format)
76		.try_collect::<Vec<_>>();
77
78	let events = services
79		.timeline
80		.pdus_rev(Some(sender_user), room_id, Some(timeline_end.saturating_add(1)))
81		.wide_and_then(|item| visibility_filter(&services, item, sender_user).map(Ok))
82		.ready_try_filter_map(Ok)
83		.try_take(limit)
84		.try_collect()
85		.map_ok(|mut vec: Vec<_>| {
86			vec.reverse();
87			vec
88		});
89
90	let account_data = services
91		.account_data
92		.changes_since_fallible(
93			Some(room_id),
94			sender_user,
95			0,
96			Some(timeline_end.into_normal().into_unsigned()),
97		)
98		.ready_try_filter_map(|e| Ok(extract_variant!(e, AnyRawAccountDataEvent::Room)))
99		.try_collect::<Vec<_>>();
100
101	let (visibility, state, events, account_data) =
102		try_join4(visibility, state, events, account_data)
103			.boxed() // erase the state stream's higher-ranked event lifetime
104			.await?;
105
106	Ok(Response {
107		room_id: room_id.to_owned(),
108		membership,
109		visibility: visibility.into(),
110		account_data: Some(account_data),
111		state: state.into(),
112		messages: PaginationChunk {
113			start: events
114				.first()
115				.map(at!(0))
116				.as_ref()
117				.map(ToString::to_string),
118
119			end: events
120				.last()
121				.map(at!(0))
122				.as_ref()
123				.map_or_else(|| timeline_end.to_string(), ToString::to_string),
124
125			chunk: events
126				.into_iter()
127				.map(at!(1))
128				.map(Event::into_format)
129				.collect(),
130		}
131		.into(),
132	})
133}
tuwunel_api_client/room/initial_sync.rs

tuwunel_api_client/room/
initial_sync.rs
