tuwunel_api/client/sync/
mod.rs

1mod v3;
2mod v5;
3
4use futures::{StreamExt, pin_mut};
5use ruma::{RoomId, UserId, events::TimelineEventType::RoomMember};
6use tuwunel_core::{
7	Error, PduCount, Result,
8	matrix::{Event, pdu::PduEvent},
9	utils::{ReadyExt, result::LogErr, stream::BroadbandExt},
10};
11use tuwunel_service::Services;
12
13pub(crate) use self::{
14	v3::{calculate_heroes, sync_events_route},
15	v5::sync_events_v5_route,
16};
17
18#[derive(Clone, Copy)]
19enum TimelineErrors {
20	Ignore,
21	Propagate,
22}
23
24async fn load_timeline(
25	services: &Services,
26	sender_user: &UserId,
27	room_id: &RoomId,
28	roomsincecount: PduCount,
29	next_batch: Option<PduCount>,
30	limit: usize,
31) -> Result<(Vec<(PduCount, PduEvent)>, bool, PduCount), Error> {
32	load_timeline_with_errors(
33		services,
34		sender_user,
35		room_id,
36		roomsincecount,
37		next_batch,
38		limit,
39		TimelineErrors::Ignore,
40	)
41	.await
42}
43
44async fn load_timeline_fallible(
45	services: &Services,
46	sender_user: &UserId,
47	room_id: &RoomId,
48	roomsincecount: PduCount,
49	next_batch: Option<PduCount>,
50	limit: usize,
51) -> Result<(Vec<(PduCount, PduEvent)>, bool, PduCount), Error> {
52	load_timeline_with_errors(
53		services,
54		sender_user,
55		room_id,
56		roomsincecount,
57		next_batch,
58		limit,
59		TimelineErrors::Propagate,
60	)
61	.await
62}
63
64async fn load_timeline_with_errors(
65	services: &Services,
66	sender_user: &UserId,
67	room_id: &RoomId,
68	roomsincecount: PduCount,
69	next_batch: Option<PduCount>,
70	limit: usize,
71	errors: TimelineErrors,
72) -> Result<(Vec<(PduCount, PduEvent)>, bool, PduCount), Error> {
73	let until = next_batch.map(|count| count.saturating_add(1));
74	let pdus = services
75		.timeline
76		.pdus_rev(Some(sender_user), room_id, until);
77
78	// Take the last events for the timeline.
79	pin_mut!(pdus);
80	let mut timeline_pdus = Vec::new();
81	let mut last_timeline_count = PduCount::max();
82	let mut first = true;
83	let mut limited = false;
84
85	while let Some(pdu) = pdus.next().await {
86		let (pducount, pdu) = match pdu {
87			| Ok(pdu) => pdu,
88			| Err(error) if first || matches!(errors, TimelineErrors::Propagate) => {
89				return Err(error);
90			},
91			| Err(_) => continue,
92		};
93
94		if first {
95			first = false;
96			last_timeline_count = matches!(pducount, PduCount::Normal(_))
97				.then_some(pducount)
98				.unwrap_or_else(PduCount::max);
99		}
100
101		if pducount <= roomsincecount {
102			break;
103		}
104
105		if timeline_pdus.len() == limit {
106			limited = true;
107			break;
108		}
109
110		timeline_pdus.push((pducount, pdu));
111	}
112
113	timeline_pdus.reverse();
114
115	Ok((timeline_pdus, limited, last_timeline_count))
116}
117
118async fn share_encrypted_room(
119	services: &Services,
120	sender_user: &UserId,
121	user_id: &UserId,
122	ignore_room: Option<&RoomId>,
123) -> bool {
124	services
125		.state_cache
126		.get_shared_rooms(sender_user, user_id)
127		.ready_filter(|&room_id| Some(room_id) != ignore_room)
128		.map(ToOwned::to_owned)
129		.broad_any(async |other_room_id| {
130			services
131				.state_accessor
132				.is_encrypted_room(&other_room_id)
133				.await
134		})
135		.await
136}
137
138/// State sections strip the stored `prev_content`/`prev_sender` pair
139/// (Synapse injects the pair on timeline fetches only). The requester's own
140/// membership and events duplicated from the returned timeline (MSC4222,
141/// full_state) keep it: clients read membership transitions from those
142/// copies.
143fn strip_prev_state(
144	mut pdu: PduEvent,
145	sender_user: &UserId,
146	in_timeline: impl Fn(&PduEvent) -> bool,
147) -> PduEvent {
148	let own_membership =
149		*pdu.kind() == RoomMember && pdu.state_key() == Some(sender_user.as_str());
150
151	if !own_membership && !in_timeline(&pdu) {
152		pdu.remove_prev_state().log_err().ok();
153	}
154
155	pdu
156}
tuwunel_api/client/sync/mod.rs

tuwunel_api/client/sync/
mod.rs
