tuwunel_api_client/dehydrated_device/
get_dehydrated_events.rs

1use axum::extract::State;
2use futures::StreamExt;
3use ruma::{
4	DeviceId, UserId,
5	api::client::dehydrated_device::get_events::{
6		unstable_v1::{Request as ByBodyRequest, Response as ByBodyResponse},
7		unstable_v2::{Request, Response},
8	},
9	events::AnyToDeviceEvent,
10	serde::Raw,
11};
12use tuwunel_core::{
13	Err, Result, err,
14	utils::{result::IsErrOr, stream::ReadyExt},
15};
16use tuwunel_service::Services;
17
18use crate::{ClientIp, Ruma};
19
20#[derive(Default)]
21struct Page {
22	events: Vec<Raw<AnyToDeviceEvent>>,
23	last: Option<u64>,
24	more: bool,
25}
26
27const MAX_BATCH_EVENTS: usize = 50;
28
29/// # `GET /_matrix/client/../dehydrated_device/{device_id}/events`
30///
31/// Paginates the events of the dehydrated device.
32///
33/// `next_batch` is omitted on the last page, as MSC3814 requires.
34#[tracing::instrument(
35	level = "debug",
36	skip_all,
37	fields(
38		%client,
39	),
40)]
41pub(crate) async fn get_dehydrated_events_route(
42	State(services): State<crate::State>,
43	ClientIp(client): ClientIp,
44	body: Ruma<Request>,
45) -> Result<Response> {
46	let since = body.from.as_deref();
47	let Page { events, last, more } =
48		dehydrated_events(&services, body.sender_user(), &body.device_id, since).await?;
49
50	let next_batch = last
51		.filter(|_| more)
52		.as_ref()
53		.map(ToString::to_string);
54
55	Ok(Response { next_batch, events })
56}
57
58/// # `POST /_matrix/client/../dehydrated_device/{device_id}/events`
59///
60/// Paginates the events of the dehydrated device.
61///
62/// Older clients still send this earlier unstable form, with the batch token
63/// in the body. An empty page ends the events.
64#[tracing::instrument(
65	level = "debug",
66	skip_all,
67	fields(
68		%client,
69	),
70)]
71pub(crate) async fn get_dehydrated_events_by_body_route(
72	State(services): State<crate::State>,
73	ClientIp(client): ClientIp,
74	body: Ruma<ByBodyRequest>,
75) -> Result<ByBodyResponse> {
76	let since = body.next_batch.as_deref();
77	let Page { events, last, .. } =
78		dehydrated_events(&services, body.sender_user(), &body.device_id, since).await?;
79
80	let next_batch = last.as_ref().map(ToString::to_string);
81
82	Ok(ByBodyResponse { next_batch, events })
83}
84
85async fn dehydrated_events(
86	services: &Services,
87	sender_user: &UserId,
88	device_id: &DeviceId,
89	since: Option<&str>,
90) -> Result<Page> {
91	let existing_id = services
92		.users
93		.get_dehydrated_device_id(sender_user)
94		.await;
95
96	if existing_id
97		.as_ref()
98		.is_err_or(|existing_id| existing_id != device_id)
99	{
100		return Err!(Request(Forbidden("Not the dehydrated device_id.")));
101	}
102
103	let since = since
104		.filter(|since| !since.is_empty())
105		.map(str::parse)
106		.transpose()
107		.map_err(|_| err!(Request(InvalidParam("Unparsable dehydrated device batch token."))))?;
108
109	let page = services
110		.users
111		.get_to_device_events(sender_user, device_id, since, None)
112		.take(MAX_BATCH_EVENTS.saturating_add(1))
113		.ready_fold(Page::default(), Page::push)
114		.await;
115
116	Ok(page)
117}
118
119impl Page {
120	fn push(mut self, (count, event): (u64, Raw<AnyToDeviceEvent>)) -> Self {
121		if self.events.len() < MAX_BATCH_EVENTS {
122			self.events.push(event);
123			self.last = Some(count);
124		} else {
125			self.more = true;
126		}
127
128		self
129	}
130}
tuwunel_api_client/dehydrated_device/get_dehydrated_events.rs

tuwunel_api_client/dehydrated_device/
get_dehydrated_events.rs
